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

fix(web): address model retry review feedback

Browse Source
This commit is contained in:
Yichen Jiang
2026-07-29 14:23:03 +08:00
parent 602f594c4b
commit 49131c47c5
28 changed files with 301 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/client/runtime/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/client/runtime/README.md
README.md: 4f4a957baa8e8e24486318d3ffde235155b5cf73
README.zh.md: 51f8bc91d59b6971d4a3ff9dfe3d0edd210d2557
README.md: 45cc032db81aee79061fc2f6a9d9062513629000
README.zh.md: a769661f6e4f7b1eb6d9872c5b979a871aba3eba

2
packages/client/runtime/README.md
View File

@@ -26,7 +26,7 @@ SlotsService gives the renderer separate bare observables for `useSessions` and
## Model retry projection
The Session object validates plugin-owned, provider-routed `llm/retry` payloads at the event wire boundary. A valid event removes the matching failed step's streaming partial and inserts a durable retry notice at the event's sequence position. Normal-mode notices carry their finite maximum; always-mode notices remain explicitly unbounded. Window rebuild and history replay apply the same projection, so logged chunks from the discarded attempt never reappear as an interrupted reply after refresh. A terminal turn without `llm/retry` retains the existing behavior: visible unfinalized output is frozen as an interrupted assistant node.
The Session object validates plugin-owned, provider-routed `llm/retry` payloads at the event wire boundary against the producer's complete field contract, including timer, integer, status, provider-delay, and non-empty diagnostic bounds. A valid event removes the matching failed step's streaming partial and inserts a durable retry notice at the event's sequence position. The notice is `scheduled` until a following retry turn starts; an aborted or disposed source turn marks it `cancelled`, while the retry turn marks it `started`. Normal-mode notices carry their finite maximum; always-mode notices remain explicitly unbounded. Window rebuild and history replay apply the same projection, so logged chunks from the discarded attempt never reappear as an interrupted reply after refresh. A terminal turn without `llm/retry` retains the existing behavior: visible unfinalized output is frozen as an interrupted assistant node.
## Session model selection

2
packages/client/runtime/README.zh.md
View File

@@ -26,7 +26,7 @@ SlotsService 分别为 renderer 提供 `useSessions` 与 `useWorkspaces` 的裸
## 模型重试投影
Session 对象会在事件 wire 边界验证由插件负责、按提供方路由的 `llm/retry` 载荷。有效事件会移除对应失败步骤的流式输出片段，并在该事件的序列位置插入一条持久的重试提示。normal mode 提示携带其有限上限；always mode 提示则保持显式无界。窗口重建与历史回放应用相同的投影，因此刷新后，来自已丢弃尝试的日志分片绝不会重新显示为中断回复。没有 `llm/retry` 的终止轮次保留现有行为：可见但尚未定稿的输出会冻结为中断的 assistant 节点。
Session 对象会在事件 wire 边界依据生产方的完整字段契约，验证由插件负责、按提供方路由的 `llm/retry` 载荷，包括计时器、整数、状态、提供方延迟和非空诊断字段的边界。有效事件会移除对应失败步骤的流式输出片段，并在该事件的序列位置插入一条持久的重试提示。该提示在后续重试轮次开始前为 `scheduled`；源轮次中止或释放会将其标记为 `cancelled`，重试轮次则会将其标记为 `started`。normal mode 提示携带其有限上限；always mode 提示则保持显式无界。窗口重建与历史回放应用相同的投影，因此刷新后，来自已丢弃尝试的日志分片绝不会重新显示为中断回复。没有 `llm/retry` 的终止轮次保留现有行为：可见但尚未定稿的输出会冻结为中断的 assistant 节点。
## 会话模型选择

1
packages/client/runtime/package.json
View File

@@ -50,6 +50,7 @@
},
"devDependencies": {
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-timeout": "workspace:^",
"@types/react": "~18.3.1",
"cordis": "^4.0.0-rc.7"
},

5
packages/client/runtime/src/client/sessions/conversation.ts
View File

@@ -97,6 +97,11 @@ export type ModelRetryNode = LlmRetryEventData & {
seq: number
/** Unix epoch ms from the llm/retry session event. */
time: number
/**
* Client-derived lifecycle: scheduled until a retry turn starts, started
* once it does, or cancelled when the failed turn aborts first.
*/
retryState: 'scheduled' | 'started' | 'cancelled'
}
/** A tool result paired (when in-window) with its call head. */

73
packages/client/runtime/src/client/sessions/session.ts
View File

@@ -13,8 +13,8 @@ import type {
import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
import type { SessionFace } from '../contract/session.ts'
import type {
CodeSubCall, ComposerPhase, ConversationNode, ConversationSnapshot, OpenState,
PromptError, QueuedMessage, RunningToolCall,
CodeSubCall, ComposerPhase, ConversationNode, ConversationSnapshot, ModelRetryNode,
OpenState, PromptError, QueuedMessage, RunningToolCall,
} from './conversation.ts'
import type { PendingInteraction } from './pending.ts'
import { PendingWait } from './pending.ts'
@@ -27,6 +27,10 @@ import type { ProjectionsBaseline } from './projection-store.ts'
/** Messages requested per history page. */
export const PAGE_MESSAGES = 50
// Browser bundles cannot value-import the host timeout library. This protocol
// bound is pinned to @deepseek-ai/dsh-timeout's MAX_TIMER_DELAY_MS in tests.
const MAX_RETRY_DELAY_MS = 2_147_483_647
/** Manager-owned observers of a Session object's local state edges. */
export interface SessionOptions {
/**
@@ -637,6 +641,7 @@ export class Session implements SessionFace {
kind: 'model-retry',
seq: event.seq,
time: event.time,
retryState: 'scheduled',
...data,
})
this.derivedRev++
@@ -702,6 +707,10 @@ export class Session implements SessionFace {
return
}
switch (event.type) {
case 'turn/start': {
if (event.data.trigger.kind === 'retry') this.settleScheduledRetry('started')
return
}
case 'assistant/chunk': {
const { turn, step, chunk } = event.data
if (this.partial === null || this.partial.turn !== turn || this.partial.step !== step) {
@@ -730,6 +739,9 @@ export class Session implements SessionFace {
return
}
case 'turn/end': {
if (event.data.reason.kind === 'aborted' || event.data.reason.kind === 'disposed') {
this.settleScheduledRetry('cancelled', event.data.turn)
}
// Aborted turns never finalize. The accumulated partial is VALUE, not residue: freeze it
// into an interrupted terminal node (pulse stops, text survives) instead of deleting it.
// Shared by live and window-replay paths, so a refresh reconstructs the same frozen node
@@ -771,6 +783,27 @@ export class Session implements SessionFace {
}
}
/**
* Settle the newest scheduled retry, optionally restricted to its failed turn.
* @param retryState - next client projection state to publish.
* @param turn - failed turn required for cancellation; omitted for the next retry turn start.
*/
private settleScheduledRetry(
retryState: Exclude<ModelRetryNode['retryState'], 'scheduled'>,
turn?: number,
): void {
const index = this.derivedNodes.findLastIndex(node =>
node.kind === 'model-retry'
&& node.retryState === 'scheduled'
&& (turn === undefined || node.turn === turn))
if (index < 0) return
const node = this.derivedNodes[index]
/* v8 ignore next -- findLastIndex's predicate narrows the indexed node only at runtime. */
if (node?.kind !== 'model-retry') return
this.derivedNodes[index] = { ...node, retryState }
this.derivedRev++
}
/** Re-derive state (partial/openCalls/derivedNodes) from raw window events after a rebuild — keeps
* paging/stitching consistent, and makes live handling and history replay converge on the same
* retry notices and interrupted nodes. */
@@ -854,37 +887,49 @@ function parseRetryEventData(value: unknown): LlmRetryEventData | null {
const failure = data.failure
if (failure === null || typeof failure !== 'object') return null
const failureData = failure as Record<string, unknown>
if (!nonNegativeInteger(data.turn)
|| !nonNegativeInteger(data.step)
if (!nonNegativeSafeInteger(data.turn)
|| !nonNegativeSafeInteger(data.step)
|| typeof data.provider !== 'string'
|| data.provider.length === 0
|| typeof data.policyKey !== 'string'
|| data.policyKey.length === 0
|| !positiveInteger(data.retry)
|| !positiveSafeInteger(data.retry)
|| typeof data.delayMs !== 'number'
|| !Number.isFinite(data.delayMs)
|| data.delayMs < 0
|| data.delayMs > MAX_RETRY_DELAY_MS
|| typeof failureData.message !== 'string'
|| typeof failureData.code !== 'string') return null
|| failureData.message.length === 0
|| typeof failureData.code !== 'string'
|| failureData.code.length === 0) return null
if (data.mode === 'normal') {
if (!positiveInteger(data.maxRetries) || data.retry > data.maxRetries) return null
if (!positiveSafeInteger(data.maxRetries) || data.retry > data.maxRetries) return null
} else if (data.mode === 'always') {
if ('maxRetries' in data) return null
} else {
return null
}
const optionalNumbers = [failureData.status, failureData.providerRetryAfterMs]
if (optionalNumbers.some(item => item !== undefined && (typeof item !== 'number' || !Number.isFinite(item)))) return null
if (failureData.requestId !== undefined && typeof failureData.requestId !== 'string') return null
if (failureData.status !== undefined
&& (typeof failureData.status !== 'number'
|| !Number.isInteger(failureData.status)
|| failureData.status < 100
|| failureData.status > 599)) return null
if (failureData.providerRetryAfterMs !== undefined
&& (typeof failureData.providerRetryAfterMs !== 'number'
|| !Number.isFinite(failureData.providerRetryAfterMs)
|| failureData.providerRetryAfterMs <= 0)) return null
if (failureData.requestId !== undefined
&& (typeof failureData.requestId !== 'string'
|| failureData.requestId.length === 0)) return null
return data as unknown as LlmRetryEventData
}
function nonNegativeInteger(value: unknown): value is number {
return typeof value === 'number' && Number.isInteger(value) && value >= 0
function nonNegativeSafeInteger(value: unknown): value is number {
return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0
}
function positiveInteger(value: unknown): value is number {
return nonNegativeInteger(value) && value > 0
function positiveSafeInteger(value: unknown): value is number {
return nonNegativeSafeInteger(value) && value > 0
}
/**

2
packages/client/runtime/tests/event-script.ts
View File

@@ -81,7 +81,7 @@ export const ev = {
failure: { code: 'TRANSPORT', message },
},
}),
turnEnd: (seq: number, turn: number, reason: 'completed' | 'cancelled' = 'completed'): SessionEvent =>
turnEnd: (seq: number, turn: number, reason: 'completed' | 'aborted' | 'disposed' = 'completed'): SessionEvent =>
at(seq, { type: 'turn/end', data: { turn, reason: { kind: reason } } }),
commandRun: (seq: number, commandId: string, name: string, args = ''): SessionEvent =>
at(seq, { type: 'command/run', data: { commandId, name, args, source: { kind: 'user' } } }),

112
packages/client/runtime/tests/session.spec.ts
View File

@@ -8,6 +8,7 @@
import { describe, expect, it, vi } from 'vitest'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
import { Session } from '../src/client/sessions/session.ts'
import { FakeApiClient, deferred, err, ok } from './fake-api.ts'
@@ -179,6 +180,7 @@ describe('live event path', () => {
expect(snapshot.partial).toBeNull()
expect(snapshot.nodes.at(-1)).toMatchObject({
kind: 'model-retry',
retryState: 'scheduled',
turn: 1,
step: 0,
provider: 'fake',
@@ -194,6 +196,7 @@ describe('live event path', () => {
for (const event of retryTurn.slice(7)) feed(event)
snapshot = session.getSnapshot()
expect(snapshot.nodes.slice(-2).map(node => node.kind)).toEqual(['model-retry', 'assistant'])
expect(snapshot.nodes.at(-2)).toMatchObject({ kind: 'model-retry', retryState: 'started' })
expect(snapshot.nodes.at(-1)).toMatchObject({ kind: 'assistant', blocks: [{ kind: 'text', text: '完整回复' }] })
const replay = makeSession()
@@ -203,31 +206,88 @@ describe('live event path', () => {
expect(replay.session.getSnapshot().partial).toBeNull()
})
it('ignores malformed retry payloads without retracting the current partial', async () => {
it('rejects retry payloads outside the producer contract without retracting the current partial', async () => {
const { session } = await opened()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
feed(ev.turnStart(6, 1))
feed(ev.chunkStart(7, 1))
feed(ev.chunkText(8, 1, '仍在生成'))
const valid = {
turn: 1, step: 0,
provider: 'fake', mode: 'normal', policyKey: 'fake-normal',
retry: 1, maxRetries: 2, delayMs: 500,
failure: { code: 'TRANSPORT', message: 'temporary failure' },
}
const invalid = [
{ ...valid, turn: Number.MAX_SAFE_INTEGER + 1 },
{ ...valid, step: Number.MAX_SAFE_INTEGER + 1 },
{ ...valid, provider: '' },
{ ...valid, policyKey: '' },
{ ...valid, retry: Number.MAX_SAFE_INTEGER + 1 },
{ ...valid, maxRetries: Number.MAX_SAFE_INTEGER + 1 },
{ ...valid, delayMs: -1 },
{ ...valid, delayMs: Number.POSITIVE_INFINITY },
{ ...valid, delayMs: MAX_TIMER_DELAY_MS + 1 },
{ ...valid, failure: { ...valid.failure, message: '' } },
{ ...valid, failure: { ...valid.failure, code: '' } },
{ ...valid, failure: { ...valid.failure, status: '429' } },
{ ...valid, failure: { ...valid.failure, status: 99 } },
{ ...valid, failure: { ...valid.failure, status: 429.5 } },
{ ...valid, failure: { ...valid.failure, status: 600 } },
{ ...valid, failure: { ...valid.failure, providerRetryAfterMs: 0 } },
{ ...valid, failure: { ...valid.failure, providerRetryAfterMs: Number.POSITIVE_INFINITY } },
{ ...valid, failure: { ...valid.failure, requestId: 1 } },
{ ...valid, failure: { ...valid.failure, requestId: '' } },
]
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
try {
feed(at(9, {
type: 'llm/retry',
data: {
turn: 1, step: 0,
provider: 'fake', mode: 'normal', policyKey: 'fake-normal',
retry: 3, maxRetries: 2, delayMs: 500,
failure: { code: 'TRANSPORT', message: 'bad budget' },
},
}))
for (const [index, data] of invalid.entries()) {
feed(at(9 + index, { type: 'llm/retry', data }))
}
expect(session.getSnapshot().partial?.blocks).toEqual([{ kind: 'text', text: '仍在生成' }])
expect(session.getSnapshot().nodes.filter(node => node.kind === 'model-retry')).toEqual([])
expect(errorSpy).toHaveBeenCalledTimes(invalid.length)
expect(errorSpy).toHaveBeenCalledWith('[web-runtime] ignored malformed llm/retry event at seq 9')
} finally {
errorSpy.mockRestore()
}
})
it('accepts complete retry payloads at the producer field boundaries', async () => {
const { session } = await opened()
session.handleMuxEnvelope('r' as never, {
type: 'session/event',
sessionId: SID,
event: at(6, {
type: 'llm/retry',
data: {
turn: Number.MAX_SAFE_INTEGER,
step: Number.MAX_SAFE_INTEGER,
provider: 'fake',
mode: 'normal',
policyKey: 'fake-normal',
retry: Number.MAX_SAFE_INTEGER,
maxRetries: Number.MAX_SAFE_INTEGER,
delayMs: MAX_TIMER_DELAY_MS,
failure: {
code: 'RATE_LIMIT',
message: 'provider busy',
status: 599,
providerRetryAfterMs: Number.MIN_VALUE,
requestId: 'req-1',
},
},
}),
})
expect(session.getSnapshot().nodes.at(-1)).toMatchObject({
kind: 'model-retry',
retryState: 'scheduled',
retry: Number.MAX_SAFE_INTEGER,
delayMs: MAX_TIMER_DELAY_MS,
failure: { status: 599, providerRetryAfterMs: Number.MIN_VALUE, requestId: 'req-1' },
})
})
it('projects always-mode retries and rejects mode-specific maximums or unknown modes', async () => {
const { session } = await opened()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
@@ -244,6 +304,7 @@ describe('live event path', () => {
}))
expect(session.getSnapshot().nodes.at(-1)).toMatchObject({
kind: 'model-retry',
retryState: 'scheduled',
mode: 'always',
retry: 3,
})
@@ -273,6 +334,27 @@ describe('live event path', () => {
}
})
it.each(['aborted', 'disposed'] as const)(
'marks a scheduled retry as cancelled when its failed turn ends %s',
async (reason) => {
const { session } = await opened()
const feed = (event: SessionEvent) => {
session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event })
}
feed(ev.turnStart(6, 1))
feed(ev.retry(7, 1))
expect(session.getSnapshot().nodes.at(-1)).toMatchObject({
kind: 'model-retry',
retryState: 'scheduled',
})
feed(ev.turnEnd(8, 1, reason))
expect(session.getSnapshot().nodes.at(-1)).toMatchObject({
kind: 'model-retry',
retryState: 'cancelled',
})
},
)
it('freezes an unfinalized partial into an interrupted node on turn/end (cancel path)', async () => {
const { session } = await opened()
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
@@ -280,7 +362,7 @@ describe('live event path', () => {
feed(ev.user(7, '要被打断的'))
feed(ev.chunkStart(8, 1))
feed(ev.chunkText(9, 1, '说到一半'))
feed(ev.turnEnd(10, 1, 'cancelled')) // no assistant/message ever arrives
feed(ev.turnEnd(10, 1, 'aborted')) // no assistant/message ever arrives
const snapshot = session.getSnapshot()
expect(snapshot.partial).toBeNull()
const frozen = snapshot.nodes.at(-1)
@@ -299,7 +381,7 @@ describe('live event path', () => {
expect(session.getSnapshot().runningCalls).toEqual([])
// Second call never resolves: turn/end freezes it as an error card.
feed(ev.toolCall(9, 1, 'c2', 'slow_tool', '{}'))
feed(ev.turnEnd(10, 1, 'cancelled'))
feed(ev.turnEnd(10, 1, 'aborted'))
const snapshot = session.getSnapshot()
expect(snapshot.runningCalls).toEqual([])
expect(snapshot.nodes.at(-1)).toMatchObject({
@@ -615,7 +697,7 @@ describe('remaining branches', () => {
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
feed(ev.turnStart(6, 1))
feed(ev.chunkStart(7, 1)) // empty text block only, no delta
feed(ev.turnEnd(8, 1, 'cancelled'))
feed(ev.turnEnd(8, 1, 'aborted'))
const snapshot = session.getSnapshot()
expect(snapshot.partial).toBeNull()
expect(snapshot.nodes.filter(n => n.kind === 'assistant' && (n as { interrupted?: true }).interrupted)).toEqual([])
@@ -629,7 +711,7 @@ describe('remaining branches', () => {
feed(ev.turnStart(6, 1))
feed(ev.toolCall(7, 1, 'turn1-call', 'echo', '{}'))
feed(ev.toolCall(8, 2, 'turn2-call', 'echo', '{}')) // stray call attributed to a later turn
feed(ev.turnEnd(9, 1, 'cancelled'))
feed(ev.turnEnd(9, 1, 'aborted'))
const snapshot = session.getSnapshot()
expect(snapshot.runningCalls.map(c => c.callId)).toEqual(['turn2-call'])
expect(snapshot.nodes.at(-1)).toMatchObject({ kind: 'tool-result', callId: 'turn1-call', isError: true })
@@ -723,7 +805,7 @@ describe('remaining branches', () => {
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
feed(ev.turnStart(6, 1))
feed(at(7, { type: 'assistant/chunk', data: { turn: 1, step: 0, chunk: { type: 'tool-call-delta', index: 0, id: 'c1', name: 'echo', argumentsDelta: '{' } } }))
feed(ev.turnEnd(8, 1, 'cancelled'))
feed(ev.turnEnd(8, 1, 'aborted'))
const frozen = session.getSnapshot().nodes.at(-1)
expect(frozen).toMatchObject({ kind: 'assistant', interrupted: true, blocks: [{ kind: 'tool-call', callId: 'c1' }] })
})