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

fix(web): address queue review feedback

Browse Source
This commit is contained in:
kingwl
2026-07-30 04:12:38 +08:00
parent ebbcfdd03b
commit 112b0fbad1
13 changed files with 154 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/host/apiproxy/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/host/apiproxy/README.md
README.md: e4a29dc288d7279366c2a28ed43573792979aa92
README.zh.md: 9f5eb0c4f84eb909d931dda8cbac764bf83aa2fe
README.md: 7129842a0cc89f5fa10c0bceec7cf0997ac71a99
README.zh.md: 8765627236ea7af9e5cd3b7029133181ef499905

2
packages/host/apiproxy/README.md
View File

@@ -16,7 +16,7 @@ Session titles ride the generic projection pair like every other domain — the
Session model routing is a session-domain contract. `session.models` returns the selected provider/model/reasoning target with provider-grouped advisory models, exact-route reasoning metadata, and provider-local lookup failures. `session.selectModel` validates the optional adapter-owned reasoning effort and replaces the complete target selected for the next prompt-assembly boundary. Catalog membership is not validation: an adapter may resolve an unlisted model, while an unavailable route or unsupported effort returns `model-unavailable`.
Pending queued input is a live control-plane contract, not session history. The gateway mirrors queued `InboxItem` occurrences from `agent/inbox/*` and broadcasts authoritative `session/queue` snapshots on every queued change and reconnect; pending steering stays outside this Web projection. `session.updateQueue` addresses one `InboxItemId`: edit replaces pending content and remove discards it. A driver claim wins races by retiring the address before admission; a later operation returns `queue-item-not-found`. The client never infers retirement from turn or status events.
Pending queued input is a live control-plane contract, not session history. The gateway mirrors queued `InboxItem` occurrences from `agent/inbox/*` and broadcasts authoritative `session/queue` snapshots on every queued change and reconnect; pending steering stays outside this Web projection. `session.updateQueue` addresses one `InboxItemId`: edit replaces pending content and remove discards it. A driver claim wins races by retiring the address before admission; a later operation returns `queue-item-not-found`. The operation queries only an attached Agent and never resumes a cold session because process-local inbox identities do not survive restart or disposal. The client never infers retirement from turn or status events.
Workspace and Session lists are separate reconnect baselines. `workspace.create` creates a unique name or adopts an existing directory, `workspace.delete` removes only the Workspace registration, `session.create` accepts an optional preallocated Session id, and `host/workspace-changed`, `host/workspace-removed`, plus `host/session-added` carry committed increments in either arrival order. Registration deletion preserves the directory and session logs; its Sessions remain in `session.list` and become Ungrouped. `SessionSummary.blank` and the `host/session-added` frame carry the derived zero-events bit: clients hide blank sessions and reuse them per workspace, flip blank on the first `host/session-status(running:true)`, and treat `session.list` as the reconnect authority; cold summaries are never blank because lazy persistence keeps never-appended sessions out of `list()`.

2
packages/host/apiproxy/README.zh.md
View File

@@ -16,7 +16,7 @@
会话模型路由属于会话领域契约。`session.models` 返回选中的提供方／模型／推理（reasoning）目标，以及按提供方分组的建议性模型、精确路由推理元数据和逐提供方查询失败记录。`session.selectModel` 校验由适配器持有的可选推理强度，并替换将在下一提示词组装边界使用的完整目标。目录成员关系不构成校验：适配器可以解析未列出的模型，而不可用路由或不受支持的推理强度会返回 `model-unavailable`。
待处理的 queued 输入属于实时控制平面契约，而非会话历史。网关镜像来自 `agent/inbox/*` 的 queued `InboxItem` 入队项，并在每次 queued 变更和重连时广播权威的 `session/queue` 快照；待处理 steering（中途引导）不进入此 Web 投影。`session.updateQueue` 通过 `InboxItemId` 寻址单个项：编辑会替换待处理内容，移除会将其丢弃。驱动器在接纳前退役寻址标识，因此认领会赢得竞态；之后的操作返回 `queue-item-not-found`。客户端绝不根据轮次或状态事件推断项已退役。
待处理的 queued 输入属于实时控制平面契约，而非会话历史。网关镜像来自 `agent/inbox/*` 的 queued `InboxItem` 入队项，并在每次 queued 变更和重连时广播权威的 `session/queue` 快照；待处理 steering（中途引导）不进入此 Web 投影。`session.updateQueue` 通过 `InboxItemId` 寻址单个项：编辑会替换待处理内容，移除会将其丢弃。驱动器在接纳前退役寻址标识，因此认领会赢得竞态；之后的操作返回 `queue-item-not-found`。该操作只查询当前已挂载的 Agent，绝不恢复冷会话，因为进程本地 inbox 标识无法在重启或资源释放后存活。客户端绝不根据轮次或状态事件推断项已退役。
Workspace 列表与 Session 列表是相互独立的重连基线。`workspace.create` 会创建唯一名称或接纳现有目录，`workspace.delete` 只移除 Workspace 注册记录，`session.create` 接受可选的预分配 Session id，`host/workspace-changed`、`host/workspace-removed` 与 `host/session-added` 则以任意到达顺序携带已提交的增量。删除注册记录会保留目录和会话日志；相关 Session 仍留在 `session.list` 中，并进入 Ungrouped。`SessionSummary.blank` 与 `host/session-added` 帧携带派生的零事件位：客户端隐藏空白会话并按 workspace 复用它们，在首个 `host/session-status(running:true)` 时翻转 blank，并以 `session.list` 作为重连权威；冷会话摘要永远不是空白：惰性持久化让从未追加过事件的会话根本不出现在 `list()` 中。

86
packages/host/apiproxy/src/api-proxy.ts
View File

@@ -9,7 +9,7 @@ import { join } from 'node:path'
import type { Context } from 'cordis'
import { installAgentLlmTarget } from '@deepseek-ai/dsh-agent'
import type {
Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, InboxItem,
Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, InboxItem, InboxItemId,
} from '@deepseek-ai/dsh-agent'
import { createUserMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import { errorChain } from '@deepseek-ai/dsh-llm'
@@ -514,6 +514,35 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
* identified message remain visible until every occurrence is claimed.
*/
const queuedMirror = new Map<SessionId, InboxItem[]>()
type UnseenQueueEvent =
| { readonly kind: 'update'; readonly item: InboxItem }
| { readonly kind: 'terminal' }
const unseenQueueEvents = new Map<SessionId, Map<InboxItemId, UnseenQueueEvent>>()
const rememberUnseen = (sessionId: SessionId, itemId: InboxItemId, event: UnseenQueueEvent): void => {
let events = unseenQueueEvents.get(sessionId)
if (events === undefined) {
events = new Map()
unseenQueueEvents.set(sessionId, events)
}
events.set(itemId, event)
// Only synchronous re-entrancy may deliver a mutation before its outer
// enqueue observer. Drop unmatched protocol-invalid observations instead
// of retaining process-local ids indefinitely.
queueMicrotask(() => {
const current = unseenQueueEvents.get(sessionId)
if (current?.get(itemId) !== event) return
current.delete(itemId)
if (current.size === 0) unseenQueueEvents.delete(sessionId)
})
}
const takeUnseen = (sessionId: SessionId, itemId: InboxItemId): UnseenQueueEvent | undefined => {
const events = unseenQueueEvents.get(sessionId)
const event = events?.get(itemId)
if (event === undefined) return undefined
events?.delete(itemId)
if (events?.size === 0) unseenQueueEvents.delete(sessionId)
return event
}
const publishQueue = (sessionId: SessionId): void => {
const items = queuedMirror.get(sessionId) ?? []
broadcast({
@@ -526,49 +555,59 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
})
}
ctx.effect(() => {
const retire = (agent: Agent, item: InboxItem): void => {
const retire = (agent: Agent, item: InboxItem): boolean => {
const entries = queuedMirror.get(agent.id)
if (entries === undefined) return
if (entries === undefined) {
rememberUnseen(agent.id, item.id, { kind: 'terminal' })
return false
}
const index = entries.findIndex(entry => entry.id === item.id)
if (index === -1) return
if (index === -1) {
rememberUnseen(agent.id, item.id, { kind: 'terminal' })
return false
}
entries.splice(index, 1)
if (entries.length === 0) queuedMirror.delete(agent.id)
publishQueue(agent.id)
return true
}
const disposers = [
ctx.on('agent/inbox/enqueue', (agent: Agent, item: InboxItem) => {
if (item.placement !== 'queued') return
const unseen = takeUnseen(agent.id, item.id)
if (unseen?.kind === 'terminal') return
let entries = queuedMirror.get(agent.id)
if (entries === undefined) {
entries = []
queuedMirror.set(agent.id, entries)
}
entries.push(item)
entries.push(unseen?.kind === 'update' ? unseen.item : item)
publishQueue(agent.id)
}),
ctx.on('agent/inbox/update', (agent: Agent, item: InboxItem) => {
const entries = queuedMirror.get(agent.id)
if (entries === undefined) return
if (entries === undefined) {
rememberUnseen(agent.id, item.id, { kind: 'update', item })
return
}
const index = entries.findIndex(entry => entry.id === item.id)
if (index === -1) return
if (index === -1) {
rememberUnseen(agent.id, item.id, { kind: 'update', item })
return
}
entries.splice(index, 1, item)
publishQueue(agent.id)
}),
ctx.on('agent/inbox/dequeue', (agent: Agent, item: InboxItem) => {
retire(agent, item)
if (retire(agent, item)) publishQueue(agent.id)
}),
ctx.on('agent/inbox/discard', (agent: Agent, items: InboxItem[]) => {
const entries = queuedMirror.get(agent.id)
if (entries === undefined) return
const ids = new Set(items.map(item => item.id))
const kept = entries.filter(entry => !ids.has(entry.id))
if (kept.length === entries.length) return
if (kept.length === 0) queuedMirror.delete(agent.id)
else queuedMirror.set(agent.id, kept)
publishQueue(agent.id)
let changed = false
for (const item of items) changed = retire(agent, item) || changed
if (changed) publishQueue(agent.id)
}),
ctx.on('session/disposed', (session: Session) => {
queuedMirror.delete(session.id)
unseenQueueEvents.delete(session.id)
}),
]
return () => { for (const dispose of disposers) dispose() }
@@ -1120,18 +1159,17 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
return ok(request, { accepted: true as const })
},
async updateQueue(request) {
updateQueue(request) {
const { sessionId, itemId, action } = request.payload
const found = await agentFor(sessionId)
if ('error' in found) return err(request, found.error)
if (found.agent.updateInbox(itemId, action) === 'not-found') {
return err(request, {
const agent = ctx.agents.get(sessionId)
if (agent === undefined || agent.updateInbox(itemId, action) === 'not-found') {
return Promise.resolve(err(request, {
code: 'queue-item-not-found',
message: 'queued item is no longer pending',
details: { itemId },
})
}))
}
return ok(request, { accepted: true as const })
return Promise.resolve(ok(request, { accepted: true as const }))
},
cancel(request) {

48
packages/host/apiproxy/tests/api-proxy-commands.spec.ts
View File

@@ -9,7 +9,7 @@ import { MessageId, freezeMessage } from '@deepseek-ai/dsh-llm'
* open-time queue snapshot.
*/
import { describe, expect, it } from 'vitest'
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry, { InboxItemId } from '@deepseek-ai/dsh-agent'
import type { Agent, InboxItem, InboxPlacement } from '@deepseek-ai/dsh-agent'
@@ -313,9 +313,55 @@ describe('session.updateQueue', () => {
{ id: 'claimed', action: { kind: 'remove' } },
])
})
it('rejects a stale occurrence without resuming a cold agent', async () => {
const ctx = await harness()
const resume = vi.spyOn(ctx.agents, 'resume')
const api = createApiProxy(ctx, DEFAULTS)
const response = await api.sessions.updateQueue({
rpcId: RpcId('q-cold'),
payload: {
sessionId: 'cold-session' as SessionId,
itemId: InboxItemId('stale-item'),
action: { kind: 'remove' },
},
})
expect(expectErr(response)).toMatchObject({ code: 'queue-item-not-found' })
expect(resume).not.toHaveBeenCalled()
})
})
describe('session/queue frames', () => {
it('folds nested mutations observed before their outer enqueue', async () => {
const ctx = await harness()
const agent = stubAgent(ctx)
const original = inboxItem('i-edit', inboxMessage('m-edit', 'before'), 'queued')
const edited = inboxItem('i-edit', inboxMessage('m-edit', 'after'), 'queued')
const removed = inboxItem('i-remove', inboxMessage('m-remove', 'remove me'), 'queued')
ctx.on('agent/inbox/enqueue', (subject, item) => {
if (subject !== agent) return
if (item.id === original.id) ctx.emit('agent/inbox/update', agent, edited)
if (item.id === removed.id) ctx.emit('agent/inbox/discard', agent, [removed])
})
const api = createApiProxy(ctx, DEFAULTS)
const live = new AbortController()
const collected = collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-mux-reentrant'), payload: {} }, live.signal), 2, live)
ctx.emit('agent/inbox/enqueue', agent, original)
ctx.emit('agent/inbox/enqueue', agent, removed)
const liveFrames = (await collected).filter(frame => frame.type === 'session/queue')
expect(liveFrames.map(frame => frame.items)).toEqual([
[{ id: edited.id, message: edited.message }],
])
const replay = new AbortController()
const replayFrames = await collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-mux-reentrant-replay'), payload: {} }, replay.signal), 2, replay)
expect(replayFrames.filter(frame => frame.type === 'session/queue')).toEqual(liveFrames)
})
it('publishes complete live snapshots and replays the latest snapshot on reconnect', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)