mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
fix(web): narrow request lifecycle fences
This commit is contained in:
Hypatia May
@@ -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: 03767fb5da46106a756207a0dda6353ae83aebc0
|
||||
README.zh.md: fe495bb99c81e80695887e9c71b8ebdfe3f7fe78
|
||||
README.md: cc3c4a8ad293451323a757a8ee85e3b903dc176f
|
||||
README.zh.md: f95b3966708b19e57b5c9ef46e0d165dafffdd53
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
Client cordis boot and React-free object services: SlotsService wraps SlotCore and supplies renderer data sources; SessionsService owns Session objects, list/scope/history state; WorkspacesService depends on SessionsService and owns Workspace objects, list/actions, default-target derivation, and the New Session blank-reuse entry (`connectWorkspace`). The runtime fans the shared Host stream into both managers. Client sessions are always Host-born (Session+Agent+cwd in one `session.create`); the client holds no pre-entity session state — a session's Agent scope (the client mirror of host dsh-scope, keyed by the shared agent/session id) is born when its row enters the list mirror and dies with the prune. Contract: api-contracts v3 §4. Each `Session` holds a generic `ProjectionValueStore` seeded from the history-tail `projections` block and updated by `session/projection` frames under higher-seq-wins; domain keys (including `todos`, `title`, and `tokenUsage`) are read via `projections.faceOf` / `useProjection`, not via `ConversationSnapshot`. `ConversationSnapshot.modelRequest` separately retains the complete latest `session/model-request` observed on the current mux connection. Each frame replaces the whole snapshot, so omitted numerator or capacity fields clear an earlier value. `SessionManager` buffers one pre-instantiation snapshot, while `session/subscribed`, disconnect, and removal clear resident and pending values; removal also installs a Host-lifecycle tombstone so a late frame from the independent mux stream cannot repopulate the removed session. Reconnect, restore, and a new subscription therefore show no context percentage until another request is observed. Model selection alone does not alter request telemetry.
|
||||
Client cordis boot and React-free object services: SlotsService wraps SlotCore and supplies renderer data sources; SessionsService owns Session objects, list/scope/history state; WorkspacesService depends on SessionsService and owns Workspace objects, list/actions, default-target derivation, and the New Session blank-reuse entry (`connectWorkspace`). The runtime fans the shared Host stream into both managers. Client sessions are always Host-born (Session+Agent+cwd in one `session.create`); the client holds no pre-entity session state — a session's Agent scope (the client mirror of host dsh-scope, keyed by the shared agent/session id) is born when its row enters the list mirror and dies with the prune. Contract: api-contracts v3 §4. Each `Session` holds a generic `ProjectionValueStore` seeded from the history-tail `projections` block and updated by `session/projection` frames under higher-seq-wins; domain keys (including `todos`, `title`, and `tokenUsage`) are read via `projections.faceOf` / `useProjection`, not via `ConversationSnapshot`. `ConversationSnapshot.modelRequest` separately retains the complete latest `session/model-request` observed on the current mux connection. Each frame replaces the whole snapshot, so omitted numerator or capacity fields clear an earlier value. `SessionManager` buffers one pre-instantiation snapshot, while `session/subscribed`, disconnect, and removal clear resident and pending values; removal also installs a request-only fence so a late transient frame from the independent mux stream cannot repopulate request telemetry, and the next mux subscription or connection generation releases that fence without blocking replayable frame classes. Reconnect, restore, and a new subscription therefore show no context percentage until another request is observed. Model selection alone does not alter request telemetry.
|
||||
|
||||
## Workspace and Session lists
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
客户端 cordis 启动与不依赖 React 的对象服务：SlotsService 包装 SlotCore 并提供 renderer 数据源；SessionsService 拥有 Session 对象、列表／scope／history 状态；WorkspacesService 依赖 SessionsService，拥有 Workspace 对象、列表／操作、默认目标派生，以及 New Session 空会话复用入口（`connectWorkspace`）。运行时把共享 Host 流分发给两个 manager。客户端 Session 一律由 Host 出生（一次 `session.create` 同瞬产出 Session+Agent+cwd）；客户端不持有任何实体化之前的会话状态——Agent scope（host dsh-scope 的客户端镜像，以 agent/session 共用 id 为键）在会话行进入列表镜像时出生，随 prune 死亡。契约：api-contracts v3 §4。每个 `Session` 持有一个通用的 `ProjectionValueStore`，由历史尾页的 `projections` 块播种，并经 `session/projection` 帧按 seq 高者胜更新；领域键（含 `todos`、`title` 与 `tokenUsage`）经 `projections.faceOf`／`useProjection` 读取，不经 `ConversationSnapshot`。`ConversationSnapshot.modelRequest` 另行保留当前 mux 连接观察到的最新完整 `session/model-request`。每个帧都会替换整个快照，因此分子或容量字段一旦缺失，就会清除先前值。`SessionManager` 会缓冲一个实例化前快照；`session/subscribed`、断开连接和移除会话则会清除常驻值与待处理值；移除还会安装 Host 生命周期删除标记，避免独立 mux 流中延迟到达的帧重新填充已移除的会话。因此，重连、恢复和新订阅都不会显示上下文百分比，直到观察到另一次请求。仅选择模型不会改变请求观测数据。
|
||||
客户端 cordis 启动与不依赖 React 的对象服务：SlotsService 包装 SlotCore 并提供 renderer 数据源；SessionsService 拥有 Session 对象、列表／scope／history 状态；WorkspacesService 依赖 SessionsService，拥有 Workspace 对象、列表／操作、默认目标派生，以及 New Session 空会话复用入口（`connectWorkspace`）。运行时把共享 Host 流分发给两个 manager。客户端 Session 一律由 Host 出生（一次 `session.create` 同瞬产出 Session+Agent+cwd）；客户端不持有任何实体化之前的会话状态——Agent scope（host dsh-scope 的客户端镜像，以 agent/session 共用 id 为键）在会话行进入列表镜像时出生，随 prune 死亡。契约：api-contracts v3 §4。每个 `Session` 持有一个通用的 `ProjectionValueStore`，由历史尾页的 `projections` 块播种，并经 `session/projection` 帧按 seq 高者胜更新；领域键（含 `todos`、`title` 与 `tokenUsage`）经 `projections.faceOf`／`useProjection` 读取，不经 `ConversationSnapshot`。`ConversationSnapshot.modelRequest` 另行保留当前 mux 连接观察到的最新完整 `session/model-request`。每个帧都会替换整个快照，因此分子或容量字段一旦缺失，就会清除先前值。`SessionManager` 会缓冲一个实例化前快照；`session/subscribed`、断开连接和移除会话则会清除常驻值与待处理值；移除还会安装仅针对请求的栅栏，避免独立 mux 流中延迟到达的瞬时帧重新填充请求遥测，下一次 mux 订阅或连接 generation 会解除该栅栏，而不会阻断可回放的帧类别。因此，重连、恢复和新订阅都不会显示上下文百分比，直到观察到另一次请求。仅选择模型不会改变请求观测数据。
|
||||
|
||||
## Workspace 与 Session 列表
|
||||
|
||||
|
||||
@@ -67,11 +67,12 @@ export class SessionManager {
|
||||
*/
|
||||
private readonly modelRequests = new Map<SessionId, ModelRequestTelemetry>()
|
||||
/**
|
||||
* Host-lifecycle tombstones. Host and mux use independent SSE streams, so a
|
||||
* frame emitted before removal can arrive after host/session-removed. Keep
|
||||
* the id fenced until a later authoritative host/session-added.
|
||||
* Removal fence for the one non-replayable mux frame. Host and mux use
|
||||
* independent SSE streams, so a request emitted before removal can arrive
|
||||
* after host/session-removed. Durable/replayed frame classes stay unfenced;
|
||||
* the next mux subscription is the same-stream proof that the id is live.
|
||||
*/
|
||||
private readonly removedSessions = new Set<SessionId>()
|
||||
private readonly removedModelRequests = new Set<SessionId>()
|
||||
/** Outstanding approval questions per session, keyed by approvalId (idempotent under mux-open
|
||||
* replays of the same requested frame). Manager-owned rather than read off Session instances
|
||||
* because the sidebar must light up for sessions never instantiated. Cleared per connection
|
||||
@@ -354,7 +355,6 @@ export class SessionManager {
|
||||
handleMuxEnvelope(envelope: RpcRequest<MuxFrame>): void {
|
||||
const frame = envelope.payload
|
||||
if (frame.type === 'stream/error') return // Controller already treats this as stream failure
|
||||
if (this.removedSessions.has(frame.sessionId)) return
|
||||
if (frame.type === 'session/projection') {
|
||||
// Finished host-computed value: land it in the resident store whether or
|
||||
// not the Session is instantiated (list rows read the 'title' key). The
|
||||
@@ -365,12 +365,14 @@ export class SessionManager {
|
||||
return
|
||||
}
|
||||
if (frame.type === 'session/model-request') {
|
||||
if (this.removedModelRequests.has(frame.sessionId)) return
|
||||
// Transient and non-replayable: retain the whole latest request until
|
||||
// lazy instantiation. Missing fields replace rather than inherit.
|
||||
const { type: _type, sessionId, ...modelRequest } = frame
|
||||
this.modelRequests.set(sessionId, modelRequest)
|
||||
}
|
||||
if (frame.type === 'session/subscribed') {
|
||||
this.removedModelRequests.delete(frame.sessionId)
|
||||
this.modelRequests.delete(frame.sessionId)
|
||||
// Rows past the host's durable baseline rode state a restart lost; drop
|
||||
// them so last-wins cannot pin a phantom value over recomputed truth.
|
||||
@@ -438,7 +440,6 @@ export class SessionManager {
|
||||
const frame = envelope.payload
|
||||
switch (frame.type) {
|
||||
case 'host/session-added': {
|
||||
this.removedSessions.delete(frame.sessionId)
|
||||
this.mergeSummary({
|
||||
sessionId: frame.sessionId, updatedAt: Date.now(), running: false, blank: frame.blank,
|
||||
...(frame.parentSessionId !== undefined ? { parentSessionId: frame.parentSessionId } : {}),
|
||||
@@ -448,7 +449,7 @@ export class SessionManager {
|
||||
return
|
||||
}
|
||||
case 'host/session-removed': {
|
||||
this.removedSessions.add(frame.sessionId)
|
||||
this.removedModelRequests.add(frame.sessionId)
|
||||
this.recordMutation({ kind: 'remove', sessionId: frame.sessionId })
|
||||
this.sessions.get(frame.sessionId)?.handleRemoved() // instance survives (resident-instance rule), only flagged in the snapshot
|
||||
this.pendingBuffers.delete(frame.sessionId) // a removed session's buffered frames must not replay on a future instantiation
|
||||
@@ -494,6 +495,7 @@ export class SessionManager {
|
||||
else this.pendingBuffers.set(sessionId, kept)
|
||||
}
|
||||
this.modelRequests.clear()
|
||||
this.removedModelRequests.clear()
|
||||
for (const session of this.sessions.values()) session.handleReconnecting()
|
||||
}
|
||||
|
||||
|
||||
@@ -511,12 +511,18 @@ export class Session implements SessionFace {
|
||||
// pane now instead of leaving it loading throughout an outage.
|
||||
this.openState = 'error'
|
||||
this.openError = {
|
||||
code: 'internal',
|
||||
message: 'connection lost while loading session history',
|
||||
details: { sessionId: this.sessionId },
|
||||
code: 'cancelled',
|
||||
message: 'session history request cancelled after connection loss',
|
||||
details: {},
|
||||
}
|
||||
changed = true
|
||||
}
|
||||
if (this.loadingOlder) {
|
||||
// The stale request's generation-fenced finally cannot clear this bit.
|
||||
// Release the paging control synchronously at the connection boundary.
|
||||
this.loadingOlder = false
|
||||
changed = true
|
||||
}
|
||||
if (this.modelRequest !== null) {
|
||||
this.modelRequest = null
|
||||
changed = true
|
||||
|
||||
@@ -136,8 +136,7 @@ describe('runtime client apply', () => {
|
||||
bench.sinks?.onDisconnected?.()
|
||||
expect(session.getSnapshot().modelRequest).toBeNull()
|
||||
|
||||
// A second failed generation does not produce another deduplicated
|
||||
// `reconnecting` state transition, but its own disconnect callback still
|
||||
// Every failed generation invokes its own disconnect callback, which
|
||||
// clears telemetry received before that generation's handshake failed.
|
||||
bench.sinks?.onMuxEnvelope?.({
|
||||
rpcId: 'request-2' as never,
|
||||
|
||||
@@ -169,6 +169,26 @@ describe('instances', () => {
|
||||
},
|
||||
})
|
||||
expect(session.getSnapshot().modelRequest).toBeNull()
|
||||
manager.handleMuxEnvelope({
|
||||
rpcId: 'resumed-subscription' as never,
|
||||
payload: { type: 'session/subscribed', sessionId: S1, lastSeq: 0 },
|
||||
})
|
||||
manager.handleMuxEnvelope({
|
||||
rpcId: 'resumed-request' as never,
|
||||
payload: {
|
||||
type: 'session/model-request',
|
||||
sessionId: S1,
|
||||
turn: 1,
|
||||
step: 1,
|
||||
provider: 'test',
|
||||
model: 'resumed',
|
||||
contextWindow: 256_000,
|
||||
},
|
||||
})
|
||||
expect(session.getSnapshot().modelRequest).toMatchObject({
|
||||
model: 'resumed',
|
||||
contextWindow: 256_000,
|
||||
})
|
||||
|
||||
manager.handleMuxEnvelope({
|
||||
rpcId: 'request-before-lazy-removal' as never,
|
||||
|
||||
@@ -833,8 +833,9 @@ describe('resync', () => {
|
||||
expect(session.getSnapshot()).toMatchObject({
|
||||
openState: 'error',
|
||||
openError: {
|
||||
code: 'internal',
|
||||
message: 'connection lost while loading session history',
|
||||
code: 'cancelled',
|
||||
message: 'session history request cancelled after connection loss',
|
||||
details: {},
|
||||
},
|
||||
})
|
||||
|
||||
@@ -843,12 +844,34 @@ describe('resync', () => {
|
||||
expect(session.getSnapshot()).toMatchObject({
|
||||
openState: 'error',
|
||||
openError: {
|
||||
code: 'internal',
|
||||
message: 'connection lost while loading session history',
|
||||
code: 'cancelled',
|
||||
message: 'session history request cancelled after connection loss',
|
||||
details: {},
|
||||
},
|
||||
})
|
||||
})
|
||||
|
||||
it('settles an in-flight older-page load when its connection generation dies', async () => {
|
||||
const { api, session } = makeSession()
|
||||
api.onHistory = () => histResponse(plainTurn(6, 1, '新问', '新答'), true)
|
||||
await session.open()
|
||||
const stale = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
|
||||
api.onHistory = () => stale.promise
|
||||
const paging = session.loadOlder()
|
||||
expect(session.getSnapshot().loadingOlder).toBe(true)
|
||||
|
||||
session.handleReconnecting()
|
||||
expect(session.getSnapshot().loadingOlder).toBe(false)
|
||||
|
||||
stale.resolve(ok({
|
||||
events: entries(plainTurn(0, 0, '旧问', '旧答')) as never[],
|
||||
hasMore: false,
|
||||
}))
|
||||
await paging
|
||||
expect(session.getSnapshot().loadingOlder).toBe(false)
|
||||
expect(session.getSnapshot().nodes.map(node => node.seq)).toEqual([7, 9])
|
||||
})
|
||||
|
||||
it('clears request telemetry on reconnect and drops a stale in-flight history response', async () => {
|
||||
const { api, session } = makeSession()
|
||||
const stale = deferred<Awaited<ReturnType<FakeApiClient['onHistory']>>>()
|
||||
|
||||
@@ -231,7 +231,8 @@ export function ChatView({
|
||||
const runningCalls = useSession(s => s.runningCalls)
|
||||
const codeDispatches = useSession(s => s.codeDispatches)
|
||||
const openState = useSession(s => s.openState)
|
||||
const openErrorMessage = useSession(s => s.openError === null ? null : `${s.openError.message}（${s.openError.code}）`)
|
||||
const openError = useSession(s => s.openError)
|
||||
const openErrorMessage = openError === null ? null : `${openError.message}（${openError.code}）`
|
||||
const hasMore = useSession(s => s.hasMore)
|
||||
const loadingOlder = useSession(s => s.loadingOlder)
|
||||
const selectedCallId = useStore(s => s.selection?.callId)
|
||||
@@ -355,7 +356,12 @@ export function ChatView({
|
||||
<div ref={listRef} className={css.scroll} onScroll={onScroll}>
|
||||
<div className={css.column}>
|
||||
{openState === 'loading' && <div className={css.hint}>载入历史…</div>}
|
||||
{openState === 'error' && <div className={css.openError}>历史加载失败：{openErrorMessage}</div>}
|
||||
{openState === 'error' && openError?.code === 'cancelled' && (
|
||||
<div className={css.hint}>连接已中断，等待重连…</div>
|
||||
)}
|
||||
{openState === 'error' && openError?.code !== 'cancelled' && (
|
||||
<div className={css.openError}>历史加载失败：{openErrorMessage}</div>
|
||||
)}
|
||||
{hasMore && (
|
||||
<div className={css.older}>
|
||||
<button type="button" disabled={loadingOlder} onClick={loadOlderAnchored}>
|
||||
|
||||
@@ -390,6 +390,17 @@ describe('ChatView', () => {
|
||||
const loading = makeHarness({ openState: 'loading' })
|
||||
const lv = render(<loading.ChatView {...loading.props} />)
|
||||
expect(lv.getByText('载入历史…')).toBeTruthy()
|
||||
const reconnecting = makeHarness({
|
||||
openState: 'error',
|
||||
openError: {
|
||||
code: 'cancelled',
|
||||
message: 'session history request cancelled after connection loss',
|
||||
details: {},
|
||||
},
|
||||
})
|
||||
const rv = render(<reconnecting.ChatView {...reconnecting.props} />)
|
||||
expect(rv.getByText('连接已中断，等待重连…')).toBeTruthy()
|
||||
expect(rv.queryByText(/session history request/)).toBeNull()
|
||||
})
|
||||
|
||||
it('pending waits leave the flow entirely — questions and approvals both take over the composer', () => {
|
||||
|
||||
Reference in New Issue
Block a user