Merge remote-tracking branch 'origin/master' into worktree-fetest

This commit is contained in:
imccyu
2026-07-30 11:51:09 +08:00
114 changed files with 1859 additions and 617 deletions

View File

@@ -12,7 +12,7 @@ export type {
WorkspaceApi, WorkspaceId, WorkspaceView,
CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
ModelReasoningEffort, ModelTarget, SessionModels,
InboxItemId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
GoalsApi, GoalRef,
} from '@deepseek-ai/dsh-host-apiproxy/api'
export type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'

View File

@@ -1150,6 +1150,11 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
)
return ok(request, { accepted: true as const })
},
updateQueue: request => err(request, {
code: 'queue-item-not-found',
message: 'fixture has no pending queue item',
details: { itemId: request.payload.itemId },
}),
cancel: (request) => {
const replay = replays.get(request.payload.sessionId)
if (replay !== undefined) {
@@ -1587,6 +1592,7 @@ export class FixtureApiClient extends AbstractApiClient {
case 'session.selectModel': return this.api.sessions.selectModel(request)
case 'session.rename': return this.api.sessions.rename(request)
case 'session.prompt': return this.api.sessions.prompt(request)
case 'session.updateQueue': return this.api.sessions.updateQueue(request)
case 'session.cancel': return this.api.sessions.cancel(request)
case 'host.describe': return this.api.host.describe(request)
case 'host.pickDirectory': return this.api.host.pickDirectory(request, new AbortController().signal)

View File

@@ -17,7 +17,7 @@ export type {
ToolCallView, ToolResultView, WorkspaceApi, WorkspaceId, WorkspaceView,
CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
ModelReasoningEffort, ModelTarget, SessionModels,
InboxItemId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
ClientRequest, ServerResponse, ServerRequest, ClientResponse, RpcMessage, RpcReceipt,
IApiClient, SessionId, SessionEvent, ContentBlock, StreamChunk,

View File

@@ -63,6 +63,7 @@ export class FakeApiClient implements IApiClient {
=> Promise<RpcResponse<{ selected: ModelTarget }>> =
payload => Promise.resolve(ok({ selected: { provider: payload.provider, model: payload.model } }))
onPrompt: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onUpdateQueue: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onCancel: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onDescribe: (payload: unknown) => Promise<RpcResponse<{ version: string; cwd: string; attachedSessions: number }>> =
() => Promise.resolve(ok({ version: '0-fake', cwd: '/f', attachedSessions: 0 }))
@@ -99,6 +100,7 @@ export class FakeApiClient implements IApiClient {
this.record('session.selectModel', payload, this.onSelectModel(payload)),
rename: (payload: unknown) => this.record('session.rename', payload, this.onRename(payload)),
prompt: (payload: unknown) => this.record('session.prompt', payload, this.onPrompt(payload)),
updateQueue: (payload: unknown) => this.record('session.updateQueue', payload, this.onUpdateQueue(payload)),
cancel: (payload: unknown) => this.record('session.cancel', payload, this.onCancel(payload)),
}

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: b51cc0276d8635ea9faa506e30246a107c1c1418
README.zh.md: 4b2248d875ae37f1b848c51a0009d2497c6b3e61
README.md: 766d8516225cd46cb1a3a80c832d1cf55e816140
README.zh.md: 9b514afca91f604b3e895187de3b5532bf22a692

View File

@@ -16,6 +16,10 @@ SlotsService gives the renderer separate bare observables for `useSessions` and
`WorkspacesService.connectWorkspace(workspaceId)` resolves the session a New Session flow lands in: it reuses the workspace's existing blank session from the list mirror (`blank && cwd == workspace.path`) or calls `session.create({workspaceId})`, returning the session id for the caller to open. `SessionSummary.blank` mirrors the host's derived empty-log bit and only ever lowers on the client: seeded by `session.list` / the `host/session-added` frame, flipped false by the first ACCEPTED local `prompt()` (on the RPC success response — acceptance proves the user message is in the host log; a rejected first prompt keeps the session blank and reusable) and by any `running: true` status frame, re-aligned by every list re-pull. List surfaces hide blank rows; the store carries every row. `SessionsService.create` accepts an optional caller-preallocated SessionId and throws `SessionCreateError` (carrying `requestedSessionId`) on failure.
## Pending queue projection
`ConversationSnapshot.queue` is the Host's authoritative transient Queue snapshot; pending steering stays outside this projection. Each row carries its `InboxItemId`, complete editable text when every content block is text, and a flattened preview. `session/queue` replaces the whole projection; reconnect buffering retains only the latest snapshot, and neither durable turn events nor running-status changes guess that an item was claimed. `Session.updateQueue()` sends edit/remove operations without optimistic mutation, so the next Host snapshot is the sole visible commit and a claim race can surface `queue-item-not-found`.
## Code Mode sub-dispatch index
`ConversationSnapshot.codeDispatches` groups a `run_code` call's sub-dispatches under their parent callId, in start order, using the native call-block shapes: a `tool/code-dispatch-start` event lands the `RunningToolCall` form (rows derive the running ring from the shape) and its `tool/code-dispatch` settlement replaces it in place with the `ToolResultNode` form, `callTime` carrying the paired start's time. A settle whose start fell outside the replay window appends directly with `callTime: null` (duration unknown — never a fabricated zero). Live mux frames and history replay build the identical index; sub-calls never join the surface `nodes` flow; per-parent array and map references are memo-stable across unrelated snapshot swaps.

View File

@@ -16,6 +16,10 @@ SlotsService 分别为 renderer 提供 `useSessions` 与 `useWorkspaces` 的裸
`WorkspacesService.connectWorkspace(workspaceId)` 解析 New Session 流程最终落入的会话:先在列表镜像中复用该 workspace 的既有空会话(`blank && cwd == workspace.path`),未命中则调用 `session.create({workspaceId})`,返回会话 id 由调用方 open。`SessionSummary.blank` 镜像主机派生的空日志位,在客户端只降不升:由 `session.list``host/session-added` 帧播种,本地首次获 Host 接受的 `prompt()`RPC 成功响应时——受理即证明用户消息已入主机日志;首讯被拒则会话保持 blank、保持可复用与任何 `running: true` 状态帧翻为 false每次列表重拉重新对齐。列表界面隐藏 blank 行store 保留全部行。`SessionsService.create` 接受可选的、由调用方预先分配的 SessionId失败时抛出 `SessionCreateError`(携带 `requestedSessionId`)。
## 待处理队列投影
`ConversationSnapshot.queue` 是 Host 提供的权威瞬态 Queue 快照;待处理 steering中途引导不进入此投影。每行都携带其 `InboxItemId`、所有内容块均为文本时的完整可编辑文本,以及扁平化预览。`session/queue` 会整体替换该投影;重连缓冲只保留最新快照,持久轮次事件和 running 状态变化都不会猜测某个项已被认领。`Session.updateQueue()` 发送编辑/移除操作,不进行乐观更新,因此下一份 Host 快照是唯一可见的提交结果,认领竞态则会返回 `queue-item-not-found`
## Code Mode 子调用索引
`ConversationSnapshot.codeDispatches` 按父调用的 callId 和启动顺序,用原生调用块形状组织一个 `run_code` 调用的子调用:`tool/code-dispatch-start` 事件落成 `RunningToolCall` 形状(行组件从该形状推导运行中的转圈状态),其 `tool/code-dispatch` 完结事件原位替换为 `ToolResultNode` 形状,`callTime` 携带成对 start 事件的时间。start 落在回放窗口之外的完结事件则直接追加,`callTime: null`耗时未知——绝不伪造零耗时。live mux 帧与历史回放构建相同的索引;子调用永不进入 surface `nodes` 流;无关快照交换不会改变每个父调用对应的数组引用和映射引用,两者均保持 memo 稳定。

View File

@@ -8,7 +8,9 @@
* dispatch) stay on the class, invisible out here.
*/
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { RpcResult, SessionId } from '@deepseek-ai/dsh-client-connection/client'
import type {
InboxItemId, QueueAction, RpcResult, SessionId,
} from '@deepseek-ai/dsh-client-connection/client'
import type { ConversationSnapshot } from '../sessions/conversation.ts'
import type { ObservableSnapshot } from './store.ts'
@@ -36,6 +38,13 @@ export interface ISession {
* @returns acceptance, or the business error (also mirrored into snapshot.promptError).
*/
prompt(content: ContentBlock[], mode: 'queue' | 'steer'): Promise<RpcResult<{ accepted: true }>>
/**
* Apply one mutation to a still-pending queue occurrence.
* @param itemId - agent-owned inbox occurrence identity.
* @param action - edit or remove operation.
* @returns acceptance, or a business/transport error.
*/
updateQueue(itemId: InboxItemId, action: QueueAction): Promise<RpcResult<{ accepted: true }>>
/**
* Cancel the running turn.
* @returns acceptance, or the business error.

View File

@@ -7,7 +7,7 @@ import type { CommandId } from '@deepseek-ai/dsh-commands/brand'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { TodoItem } from '@deepseek-ai/dsh-session/types'
import type {
RpcError, SessionId, ToolCallView, ToolResultView,
InboxItemId, RpcError, SessionId, ToolCallView, ToolResultView,
} from '@deepseek-ai/dsh-client-connection/client'
import type { PendingInteraction } from './pending.ts'
export type { TodoItem }
@@ -216,10 +216,12 @@ export interface RunningToolCall {
}
/** One queued-message row mirrored from `session/queued` frames (key: the enqueueing prompt's rpcId when wire-sourced). */
/** One independently addressable row from the transient queue snapshot. */
export interface QueuedMessage {
readonly key: string
readonly id: InboxItemId
readonly preview: string
/** Complete editable text; null when the message contains non-text blocks. */
readonly text: string | null
}
/** In-progress assistant output (chunk accumulator product). */
@@ -277,7 +279,7 @@ export interface ConversationSnapshot {
*/
codeDispatches: ReadonlyMap<string, readonly CodeSubCall[]>
pending: readonly PendingInteraction[]
/** Read-only inbox mirror (session/queued frames + mux-open baseline; cleared by the leave-running flip). */
/** Authoritative transient inbox snapshot, replaced after every host-side change. */
queue: readonly QueuedMessage[]
running: boolean
/** Input-area shape (see {@link ComposerPhase}); derived here, switched on by consumers. */

View File

@@ -351,14 +351,14 @@ export class SessionManager {
// them so last-wins cannot pin a phantom value over recomputed truth.
this.projectionStores.get(frame.sessionId)?.truncate(frame.lastSeq)
this.notifier.markDirty()
// New mux-generation baseline: buffered session/queued frames belong to
// New mux-generation baseline: buffered session/queue frames belong to
// the previous generation and the host is about to resend the live
// snapshot — drop them, or every reconnect appends a duplicate batch
// (and enough reconnects push real approval/question frames past the
// cap). Same re-baseline signal Session uses for its own mirror.
const buffered = this.pendingBuffers.get(frame.sessionId)
if (buffered !== undefined) {
const kept = buffered.filter(item => item.payload.type !== 'session/queued')
const kept = buffered.filter(item => item.payload.type !== 'session/queue')
if (kept.length !== buffered.length) {
if (kept.length === 0) this.pendingBuffers.delete(frame.sessionId)
else this.pendingBuffers.set(frame.sessionId, kept)
@@ -383,7 +383,7 @@ export class SessionManager {
}
const session = this.sessions.get(frame.sessionId)
if (session === undefined) {
// Approval/question/queued frames never hit history: buffer for replay on
// Approval/question/queue frames never hit history: buffer for replay on
// instantiation; everything else drops (not instantiated — history fully
// backfills on open).
switch (frame.type) {
@@ -391,8 +391,12 @@ export class SessionManager {
case 'approval/resolved':
case 'question/requested':
case 'question/resolved':
case 'session/queued': {
case 'session/queue': {
const buffer = this.pendingBuffers.get(frame.sessionId) ?? []
const prior = frame.type === 'session/queue'
? buffer.findIndex(item => item.payload.type === 'session/queue')
: -1
if (prior !== -1) buffer.splice(prior, 1)
buffer.push(envelope)
if (buffer.length > PENDING_BUFFER_CAP) buffer.splice(0, buffer.length - PENDING_BUFFER_CAP)
this.pendingBuffers.set(frame.sessionId, buffer)

View File

@@ -4,8 +4,8 @@ import type { Context } from 'cordis'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type {
HistoryEntry, IApiClient, MuxFrame, RpcError, RpcId, RpcResult,
SessionId, ToolEventView,
HistoryEntry, IApiClient, InboxItemId, MuxFrame, QueueAction, RpcError,
RpcId, RpcResult, SessionId, ToolEventView,
} from '@deepseek-ai/dsh-client-connection/client'
// Value import from the inline-safe wire layer (not the connection plugin):
// plugin-to-plugin value imports are a bundle purity error.
@@ -48,14 +48,6 @@ export interface SessionOptions {
/** Queue-row preview cap: the dock renders one line, the full content never leaves the host mirror. */
const QUEUE_PREVIEW_CHARS = 200
/** Internal inbox-mirror entry: the snapshot row plus the retirement-matching fields the frames carry. */
interface QueuedEntry {
row: QueuedMessage
steering: boolean
/** JSON-serialized MessageSource (steering retirement matches by source, the host-mirror precedent). */
sourceJson: string
}
/** Single-line queue-row preview: text blocks flattened, non-text as tags, capped by code point. */
function queuePreviewOf(content: readonly ContentBlock[]): string {
const flat = content
@@ -65,6 +57,12 @@ function queuePreviewOf(content: readonly ContentBlock[]): string {
return chars.length > QUEUE_PREVIEW_CHARS ? `${chars.slice(0, QUEUE_PREVIEW_CHARS).join('')}` : flat
}
/** Recover complete composer text only when editing cannot discard non-text blocks. */
function queueTextOf(content: readonly ContentBlock[]): string | null {
if (!content.every(block => block.type === 'text')) return null
return content.map(block => block.text).join('')
}
/**
* Owns a session's event window, derived conversation state, and observable
* snapshot. React bindings remain outside this data layer. Features see only
@@ -102,9 +100,8 @@ export class Session implements SessionFace {
private callsCache: { rev: number; value: RunningToolCall[] } | null = null
private pendingRev = 0
private pendingCache: { rev: number; value: PendingInteraction[] } | null = null
/** Inbox mirror (session/queued frames + mux-open baseline). Queue frames never hit history,
* so this is stream-only state: reconnect clears it and the fresh baseline re-populates. */
private queued: QueuedEntry[] = []
/** Authoritative stream-only inbox snapshot; pending work never hits history. */
private queued: QueuedMessage[] = []
private queueRev = 0
private queueCache: { rev: number; value: QueuedMessage[] } | null = null
private frozenRev = 0
@@ -234,6 +231,15 @@ export class Session implements SessionFace {
return result
}
/** Apply one operation to a still-pending queue occurrence. */
async updateQueue(itemId: InboxItemId, action: QueueAction): Promise<RpcResult<{ accepted: true }>> {
try {
return (await this.api.sessions.updateQueue({ sessionId: this.sessionId, itemId, action })).result
} catch (error) {
return transportError(error)
}
}
/**
* Stop: contract session.cancel 1:1; failures land in promptError (same error-strip display slot).
* @returns the cancel result.
@@ -393,20 +399,15 @@ export class Session implements SessionFace {
handleMuxEnvelope(rpcId: RpcId, frame: MuxFrame): void {
switch (frame.type) {
case 'session/event': {
this.retireQueued(frame.event)
this.acceptLiveEvent(frame.event, frame.view)
return
}
case 'session/queued': {
const message = frame.message
// Row key: the enqueueing prompt's rpcId when it rode this wire (the
// provisional-echo reconciliation key); otherwise the frame envelope id.
const key = 'rpcId' in message.source ? String(message.source.rpcId) : `f:${rpcId}`
this.queued.push({
row: { key, preview: queuePreviewOf(message.content) },
steering: frame.steering,
sourceJson: JSON.stringify(message.source),
})
case 'session/queue': {
this.queued = frame.items.map(item => ({
id: item.id,
preview: queuePreviewOf(item.message.content),
text: queueTextOf(item.message.content),
}))
this.queueRev++
this.notifier.markDirty()
return
@@ -459,15 +460,6 @@ export class Session implements SessionFace {
* @param running - the new running state.
*/
handleRunning(running: boolean): void {
// Leave-running sweep (host queuedMirror precedent): discard paths (cancel,
// terminal steering drop) have no per-entry frame, so ANY not-running signal
// with a nonempty mirror clears it — checked before the equality return so a
// stale replay on an already-idle session still sweeps.
if (!running && this.queued.length > 0) {
this.queued = []
this.queueRev++
this.notifier.markDirty()
}
// Turn-start conversion: a blank session never runs, so the first
// running:true proves another端's first message landed (设计稿 2.2).
if (running && this.blankBit) {
@@ -632,27 +624,6 @@ export class Session implements SessionFace {
}
}
/** Consumption-event retirement, mirroring the host queuedMirror rules: a message-triggered
* turn/start claims the oldest non-steering entry; a steering/message drains the oldest
* steering entry with the same source (loop-authored steering matches nothing and drops none). */
private retireQueued(event: SessionEvent): void {
if (this.queued.length === 0) return
let index = -1
if (event.type === 'turn/start') {
if (event.data.trigger.kind !== 'message') return
index = this.queued.findIndex(entry => !entry.steering)
} else if (event.type === 'steering/message') {
const source = JSON.stringify(event.data.message.source)
index = this.queued.findIndex(entry => entry.steering && entry.sourceJson === source)
} else {
return
}
if (index < 0) return
this.queued.splice(index, 1)
this.queueRev++
this.notifier.markDirty()
}
/** Per-event side effects (right column of the §A.9 dispatch table):
* chunk accumulation / partial clear on finalize / openCalls add-remove. */
private applyEventSideEffects(event: SessionEvent, view?: ToolEventView): void {
@@ -832,7 +803,7 @@ export class Session implements SessionFace {
this.dispatchesCache = { rev: this.dispatchesRev, value: new Map(this.codeDispatches) }
}
if (this.queueCache === null || this.queueCache.rev !== this.queueRev) {
this.queueCache = { rev: this.queueRev, value: this.queued.map(entry => entry.row) }
this.queueCache = { rev: this.queueRev, value: this.queued }
}
const partial = this.partial?.toPartial() ?? null
return {

View File

@@ -81,6 +81,7 @@ export class FakeApiClient implements IApiClient {
Promise<RpcResponse<{ selected: ModelTarget }>> =
payload => Promise.resolve(ok({ selected: { provider: payload.provider, model: payload.model } }))
onPrompt: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onUpdateQueue: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onCancel: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
onDescribe: (payload: unknown) => Promise<RpcResponse<{ version: string; cwd: string; attachedSessions: number }>> =
@@ -118,6 +119,7 @@ export class FakeApiClient implements IApiClient {
this.record('session.selectModel', payload, this.onSelectModel(payload)),
rename: (payload: unknown) => this.record('session.rename', payload, this.onRename(payload)),
prompt: (payload: unknown) => this.record('session.prompt', payload, this.onPrompt(payload)),
updateQueue: (payload: unknown) => this.record('session.updateQueue', payload, this.onUpdateQueue(payload)),
cancel: (payload: unknown) => this.record('session.cancel', payload, this.onCancel(payload)),
}

View File

@@ -1,32 +1,41 @@
/**
* Queue mirror semantics (web input-triggers queue cut 1): session/queued
* intake, host-rule retirement (message turn/start claims oldest non-steering;
* steering/message drains by source), leave-running sweep, reconnect reset,
* pre-instantiation buffering, and snapshot reference stability.
* Queue snapshot semantics: authoritative replacement after every host-side
* change, reconnect re-baselining, pre-instantiation buffering, editable-text
* projection, and snapshot reference stability.
*/
import { describe, expect, it } from 'vitest'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { MuxFrame, RpcId, SessionId } from '@deepseek-ai/dsh-client-connection/client'
import type {
InboxItemId, MuxFrame, RpcId, SessionId,
} from '@deepseek-ai/dsh-client-connection/client'
import { Session } from '../src/client/sessions/session.ts'
import { SessionManager } from '../src/client/sessions/manager.ts'
import { FakeApiClient } from './fake-api.ts'
import { ev } from './event-script.ts'
const SID = 'fk-q1' as SessionId
const text = (t: string): ContentBlock[] => [{ type: 'text', text: t }]
const text = (value: string): ContentBlock[] => [{ type: 'text', text: value }]
const rid = (id: string): RpcId => id as RpcId
const iid = (id: string): InboxItemId => id as InboxItemId
/** session/queued frame with the wire-sourced rpcId key (the host prompt path). */
function queuedFrame(body: string, rpcId: string, steering = false): MuxFrame {
interface QueueFixture {
id: string
body: string
content?: ContentBlock[]
}
/** Build one authoritative queue snapshot. */
function queueFrame(items: QueueFixture[]): MuxFrame {
return {
type: 'session/queued',
type: 'session/queue',
sessionId: SID,
message: createUserMessage({
content: text(body),
source: { kind: 'user', rpcId: rid(rpcId) } as never,
}),
steering,
items: items.map(item => ({
id: iid(item.id),
message: createUserMessage({
content: item.content ?? text(item.body),
source: { kind: 'user', rpcId: rid(`rpc-${item.id}`) } as never,
}),
})),
}
}
@@ -34,201 +43,131 @@ function makeSession(): Session {
return new Session(SID, new FakeApiClient())
}
describe('queue intake', () => {
it('lands a queued frame as a row keyed by the source rpcId with a flat preview', () => {
describe('queue snapshot intake', () => {
it('projects stable ids, flat previews, and complete text', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-1'), queuedFrame('第一条 排队\n消息', 'p-1'))
const queue = session.getSnapshot().queue
expect(queue).toEqual([{ key: 'p-1', preview: '第一条 排队 消息' }])
session.handleMuxEnvelope(rid('env-1'), queueFrame([
{ id: 'q-1', body: '第一条 排队\n消息' },
]))
expect(session.getSnapshot().queue).toEqual([
{ id: 'q-1', preview: '第一条 排队 消息', text: '第一条 排队\n消息' },
])
})
it('falls back to the envelope rpcId when the source carries none, and tags non-text blocks', () => {
it('marks mixed-content messages non-editable while retaining their preview', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-2'), {
type: 'session/queued',
sessionId: SID,
message: createUserMessage({
content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never],
source: { kind: 'plugin', plugin: 'loop' },
}),
steering: false,
})
expect(session.getSnapshot().queue).toEqual([{ key: 'f:env-2', preview: 'hi [image]' }])
session.handleMuxEnvelope(rid('env-2'), queueFrame([{
id: 'q-image',
body: '',
content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never],
}]))
expect(session.getSnapshot().queue).toEqual([
{ id: 'q-image', preview: 'hi [image]', text: null },
])
})
it('caps the preview at 200 code points with an ellipsis', () => {
it('caps previews at 200 code points and preserves the full editable text', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-3'), queuedFrame('长'.repeat(201), 'p-cap'))
const preview = session.getSnapshot().queue[0]?.preview ?? ''
expect(Array.from(preview)).toHaveLength(201) // 200 + …
expect(preview.endsWith('')).toBe(true)
const body = '长'.repeat(201)
session.handleMuxEnvelope(rid('env-3'), queueFrame([{ id: 'q-cap', body }]))
const row = session.getSnapshot().queue[0]
expect(Array.from(row?.preview ?? '')).toHaveLength(201)
expect(row?.preview.endsWith('…')).toBe(true)
expect(row?.text).toBe(body)
})
it('replaces content, order, and membership from each authoritative frame', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-4'), queueFrame([
{ id: 'q-1', body: 'one' },
{ id: 'q-2', body: 'two' },
]))
session.handleMuxEnvelope(rid('env-5'), queueFrame([
{ id: 'q-2', body: 'two edited' },
]))
expect(session.getSnapshot().queue).toEqual([
{ id: 'q-2', preview: 'two edited', text: 'two edited' },
])
session.handleMuxEnvelope(rid('env-6'), queueFrame([]))
expect(session.getSnapshot().queue).toEqual([])
})
it('keeps the queue array reference stable across unrelated snapshot swaps', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('env-4'), queuedFrame('稳定', 'p-s'))
session.handleMuxEnvelope(rid('env-7'), queueFrame([{ id: 'q-stable', body: '稳定' }]))
const before = session.getSnapshot().queue
session.handleAgentError('unrelated') // dirties the snapshot without touching the queue
session.handleAgentError('unrelated')
expect(session.getSnapshot().queue).toBe(before)
})
})
describe('queue retirement (host queuedMirror rules)', () => {
it('a message-triggered turn/start claims the oldest non-steering row', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('先', 'p-1'))
session.handleMuxEnvelope(rid('e2'), queuedFrame('后', 'p-2'))
session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: ev.turnStart(0, 0) })
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-2'])
})
describe('queue operation transport', () => {
it('addresses the session.updateQueue RPC without optimistic local mutation', async () => {
const api = new FakeApiClient()
const session = new Session(SID, api)
session.handleMuxEnvelope(rid('env-op'), queueFrame([{ id: 'q-op', body: 'pending' }]))
const before = session.getSnapshot().queue
it('an injection-triggered turn/start claims nothing', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('留', 'p-1'))
const injection = {
...ev.turnStart(0, 0),
data: { turn: 0, trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'x' } } },
} as never
session.handleMuxEnvelope(rid('e2'), { type: 'session/event', sessionId: SID, event: injection })
expect(session.getSnapshot().queue).toHaveLength(1)
})
it('steering/message drains the source-matched steering row only', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('普通', 'p-1')) // idle → non-steering
session.handleMuxEnvelope(rid('e3'), queuedFrame('插话', 'p-2', true))
// Loop-authored steering (different source) must not consume the user entry.
const foreignSteering = {
seq: 0, time: 1,
type: 'steering/message', surfaceOp: 'append',
data: {
turn: 0,
message: createUserMessage({
content: text('loop'),
source: { kind: 'plugin', plugin: 'loop' },
}),
},
} as never
session.handleMuxEnvelope(rid('e4'), { type: 'session/event', sessionId: SID, event: foreignSteering })
expect(session.getSnapshot().queue).toHaveLength(2)
const matchedSteering = {
seq: 1, time: 2,
type: 'steering/message', surfaceOp: 'append',
data: {
turn: 0,
message: createUserMessage({
content: text('插话'),
source: { kind: 'user', rpcId: rid('p-2') },
}),
},
} as never
session.handleMuxEnvelope(rid('e5'), { type: 'session/event', sessionId: SID, event: matchedSteering })
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-1'])
})
it('a leave-running flip sweeps the whole mirror (cancel/terminal-drop cover)', () => {
const session = makeSession()
session.handleRunning(true)
session.handleMuxEnvelope(rid('e1'), queuedFrame('一', 'p-1'))
session.handleMuxEnvelope(rid('e2'), queuedFrame('二', 'p-2'))
session.handleRunning(false)
expect(session.getSnapshot().queue).toEqual([])
})
it('a stale not-running relay on an idle session still sweeps replayed rows', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('孤儿', 'p-1'))
session.handleRunning(false) // running already false: equality path must not skip the sweep
expect(session.getSnapshot().queue).toEqual([])
await expect(session.updateQueue(iid('q-op'), { kind: 'edit', content: text('next') }))
.resolves.toEqual({ ok: true, value: { accepted: true } })
expect(api.callsOf('session.updateQueue')).toEqual([{
sessionId: SID,
itemId: 'q-op',
action: { kind: 'edit', content: text('next') },
}])
expect(session.getSnapshot().queue).toBe(before)
})
})
describe('queue reconnect semantics', () => {
it('session/subscribed re-baselines the mirror: stale rows drop, the following snapshot lands fresh', () => {
it('session/subscribed clears stale state before the fresh snapshot lands', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), queuedFrame('旧连接', 'p-old'))
// New mux generation: subscribed arrives first on the same stream...
session.handleMuxEnvelope(rid('e1'), queueFrame([{ id: 'q-old', body: '旧连接' }]))
session.handleMuxEnvelope(rid('e2'), { type: 'session/subscribed', sessionId: SID, lastSeq: 10 })
expect(session.getSnapshot().queue).toEqual([])
// ...then the queue snapshot replays the live inbox.
session.handleMuxEnvelope(rid('e3'), queuedFrame('新基线', 'p-new'))
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-new'])
session.handleMuxEnvelope(rid('e3'), queueFrame([{ id: 'q-new', body: '新基线' }]))
expect(session.getSnapshot().queue.map(row => row.id)).toEqual(['q-new'])
})
it('resync must NOT clear the mirror (regression: onConnected races the mux baseline)', async () => {
it('resync does not clear a baseline that raced ahead of the host connection signal', async () => {
const session = makeSession()
// Reconnect ordering that broke: mux opened first and already delivered
// the fresh generation's baseline; host stream (and with it onConnected →
// resync) lands after. The host never resends — clearing here left the
// dock empty until the next enqueue.
session.handleMuxEnvelope(rid('e1'), { type: 'session/subscribed', sessionId: SID, lastSeq: 5 })
session.handleMuxEnvelope(rid('e2'), queuedFrame('新基线', 'p-fresh'))
session.handleMuxEnvelope(rid('e2'), queueFrame([{ id: 'q-fresh', body: '新基线' }]))
await session.resync()
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-fresh'])
expect(session.getSnapshot().queue.map(row => row.id)).toEqual(['q-fresh'])
})
it('replayed steering retires without a replayed turn/start', () => {
it('running-status changes never guess at queue retirement', () => {
const session = makeSession()
session.handleMuxEnvelope(rid('e1'), { type: 'session/subscribed', sessionId: SID, lastSeq: 5 })
session.handleMuxEnvelope(rid('e2'), queuedFrame('重连插话', 'p-steer', true))
const committed = {
seq: 6, time: 2,
type: 'steering/message', surfaceOp: 'append',
data: {
turn: 1,
message: createUserMessage({
content: text('重连插话'),
source: { kind: 'user', rpcId: rid('p-steer') },
}),
},
} as never
session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: committed })
expect(session.getSnapshot().queue).toEqual([])
session.handleMuxEnvelope(rid('e1'), queueFrame([{ id: 'q-live', body: '保留' }]))
session.handleRunning(true)
session.handleRunning(false)
expect(session.getSnapshot().queue.map(row => row.id)).toEqual(['q-live'])
})
})
describe('manager buffering of queued frames', () => {
it('buffers session/queued for uninstantiated sessions and replays before the running sync', () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
manager.handleMuxEnvelope({ rpcId: rid('b1'), payload: queuedFrame('预热', 'p-b1') })
// Instantiation replays the buffer; no summary exists, so no running sweep runs.
const session = manager.get(SID)
expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-b1'])
// The buffer is consumed: a second get must not double-replay.
expect(manager.get(SID).getSnapshot().queue).toHaveLength(1)
describe('manager buffering of queue snapshots', () => {
it('replays only the latest snapshot for an uninstantiated session', () => {
const manager = new SessionManager(new FakeApiClient())
manager.handleMuxEnvelope({ rpcId: rid('b1'), payload: queueFrame([{ id: 'q-old', body: '旧' }]) })
manager.handleMuxEnvelope({ rpcId: rid('b2'), payload: queueFrame([{ id: 'q-new', body: '新' }]) })
expect(manager.get(SID).getSnapshot().queue.map(row => row.id)).toEqual(['q-new'])
})
it('a not-running list summary sweeps replayed rows at instantiation', async () => {
const api = new FakeApiClient()
api.onList = () => Promise.resolve(ok([{ sessionId: SID, updatedAt: 1, running: false }]))
const manager = new SessionManager(api)
await manager.refreshList()
manager.handleMuxEnvelope({ rpcId: rid('b2'), payload: queuedFrame('该扫掉', 'p-b2') })
expect(manager.get(SID).getSnapshot().queue).toEqual([])
})
it('subscribed re-baselines the uninstantiated buffer: stale queued frames drop, non-queue frames survive (regression: reconnect duplication)', () => {
const api = new FakeApiClient()
const manager = new SessionManager(api)
// Generation 1 baseline lands while the session is uninstantiated, along
// with a pending approval (never re-derivable from history).
manager.handleMuxEnvelope({ rpcId: rid('g1a'), payload: queuedFrame('第一代', 'p-g1') })
it('subscribed drops the prior-generation snapshot while preserving answerable frames', () => {
const manager = new SessionManager(new FakeApiClient())
manager.handleMuxEnvelope({ rpcId: rid('g1a'), payload: queueFrame([{ id: 'q-g1', body: '第一代' }]) })
manager.handleMuxEnvelope({
rpcId: rid('g1b'),
payload: { type: 'approval/requested', sessionId: SID, approvalId: 'ap-1' as never, toolName: 'bash' },
})
// Reconnect: generation 2 replays subscribed + the SAME live queue entry.
manager.handleMuxEnvelope({ rpcId: rid('g2a'), payload: { type: 'session/subscribed', sessionId: SID, lastSeq: 3 } })
manager.handleMuxEnvelope({ rpcId: rid('g2b'), payload: queuedFrame('第一代', 'p-g1') })
manager.handleMuxEnvelope({
rpcId: rid('g2a'),
payload: { type: 'session/subscribed', sessionId: SID, lastSeq: 3 },
})
manager.handleMuxEnvelope({ rpcId: rid('g2b'), payload: queueFrame([{ id: 'q-g2', body: '第二代' }]) })
const snapshot = manager.get(SID).getSnapshot()
// One queue row (no duplicate batch); the approval survived the re-baseline.
expect(snapshot.queue.map(r => r.key)).toEqual(['p-g1'])
expect(snapshot.pending.map(p => p.kind)).toEqual(['approval'])
expect(snapshot.queue.map(row => row.id)).toEqual(['q-g2'])
expect(snapshot.pending.map(pending => pending.kind)).toEqual(['approval'])
})
})
/** ok wrapper with a typed items payload (the shared helper pins value to never[]). */
function ok(items: { sessionId: SessionId; updatedAt: number; running: boolean }[]) {
return { rpcId: rid(`ok-${items.length}`), result: { ok: true as const, value: { items: items as never[] } } }
}

View File

@@ -84,6 +84,14 @@ export class FixtureSession implements SessionFace {
throw new Error(`test session "${this.sessionId}": prompt is not stubbed — supply it on the fixture's session face`)
}
/**
* Fail-loud stub; supply `updateQueue` on the fixture's session face to exercise it.
* @returns never — always throws.
*/
updateQueue(): never {
throw new Error(`test session "${this.sessionId}": updateQueue is not stubbed — supply it on the fixture's session face`)
}
/**
* Fail-loud stub; supply `cancel` on the fixture's session face to exercise it.
* @returns never — always throws.

View File

@@ -467,6 +467,7 @@ describe('fixture session face', () => {
await runtime.sessions.add({ id: 's1' })
const bare = runtime.sessions.behavior('s1')
expect(() => bare.prompt()).toThrow(/prompt is not stubbed/)
expect(() => bare.updateQueue()).toThrow(/updateQueue is not stubbed/)
expect(() => bare.cancel()).toThrow(/cancel is not stubbed/)
expect(() => bare.command()).toThrow(/command is not stubbed/)
expect(() => bare.loadOlder()).toThrow(/loadOlder is not stubbed/)

View File

@@ -128,6 +128,7 @@ export function clientBundle(id: string, libEntry: readonly string[]): UserConfi
async load(virtualId: string) {
if (!virtualId.startsWith(CSS_VIRTUAL_PREFIX)) return null
const fileId = virtualId.slice(CSS_VIRTUAL_PREFIX.length, -CSS_VIRTUAL_SUFFIX.length)
// The virtual id otherwise hides the physical stylesheet from Rolldown's watch graph.
this.addWatchFile(fileId)
const source = await readFile(fileId)
const { code, exports: cssExports } = transform({

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/ui-conversation/README.md
README.md: 09adb3fd7504b6e79402e3eb220ec474d76c182f
README.zh.md: d92f2ca72764faae767f0f4892037af2c574d8de
README.md: 75c181f2e1240f753d1f8c30152d2978151b059f
README.zh.md: 6dc167c63af27dfbc37c51b9a55087cf2c50dc7f

View File

@@ -40,3 +40,5 @@ None; this package neither assembles nor sends a provider request.
- **The sparkle icon for the others tool row is a hand-drawn approximation** — the design glyph's vector geometry is not exportable locally; promotion into ui-primitives waits on an exact export.
- **The approval panel's "Always allow this type" is deferred** — durable grants need a grant-storage design; only allow-once/reject answer today.
- **TodoPanel truncates long item text to one ellipsized line** — the figma strip has no wrap or expand affordance; full text is not readable inline.
- **Queue edit is text-only** — rows containing non-text blocks still show a flattened preview, but their edit control is disabled because the inline editor cannot preserve those blocks. A text row's edit mode replaces delete with save and cancel; Enter saves and Escape cancels. QueueDock exposes no send-now control.
- **Web exposes pending Queue only** — the Host omits pending steering from the Queue snapshot until steering has its own interaction. A consumed `steering/message` still renders in the durable transcript so external steering remains truthful on replay.

View File

@@ -40,3 +40,5 @@ todo 两个面就是在该形状上的两个注册项,都是普通注册方插
- **others 工具行的闪光图标是手绘近似版本**:无法在本地导出设计字形的矢量几何;等到存在精确导出后再将其提升到 ui-primitives。
- **审批面板的「始终允许此类」暂缓**:持久授权需要授权存储设计;今天只能回答允许一次/拒绝。
- **TodoPanel 将过长条目截成单行省略号**figma 条没有换行或展开入口,完整文本无法在行内读完。
- **Queue 编辑仅支持文本**包含非文本块的行仍显示扁平化预览但由于内联编辑器无法保留这些块其编辑控件会被禁用。文本行进入编辑模式后删除会替换为保存和取消Enter 保存Escape 取消。QueueDock 不提供立即发送控件。
- **Web 仅暴露待处理 Queue**:在 steering中途引导拥有专用交互之前Host 不会把待处理 steering 纳入 Queue 快照。已消费的 `steering/message` 仍会渲染到持久 transcript文本记录因此从外部提交的 steering 在回放时仍能如实呈现。

View File

@@ -0,0 +1,13 @@
/** Queue contracts derived from the runtime session face and snapshot. */
import type {
ConversationSnapshot, SessionFace,
} from '@deepseek-ai/dsh-client-runtime/client'
/** One address accepted by the runtime session's queue mutation verb. */
export type QueueItemId = Parameters<SessionFace['updateQueue']>[0]
/** One mutation accepted by the runtime session's queue mutation verb. */
export type QueueAction = Parameters<SessionFace['updateQueue']>[1]
/** One row projected by the runtime session's authoritative queue snapshot. */
export type QueueRow = ConversationSnapshot['queue'][number]

View File

@@ -10,6 +10,7 @@ import type {
ArbitrateKey, ArbitrateOutcome, CommandClaim, ConsumeTokenRequest, PickOutcome,
ReferenceInsert, SubmitOutcome, TokenSpan,
} from '@deepseek-ai/dsh-client-ui-slash/client'
import type { QueueRow } from '../contract/queue.ts'
/**
* The scoped-event application verbs: the hub's bail listeners call these,
@@ -99,12 +100,8 @@ export interface ComposerKeyboard {
dismissPopup(): void
}
/** One queued-message row projected from the session/queued frames (T9 supplies the store). */
export interface QueuedMessage {
/** Stable row key: the enqueueing prompt's rpcId. */
readonly key: string
readonly preview: string
}
/** One independently addressable row projected from the transient queue snapshot. */
export type QueuedMessage = QueueRow
/** Guard union of the scoped consume-token event, checked by the machine. */
export type ConsumeTokenGuard = ConsumeTokenRequest['guard']

View File

@@ -1,30 +1,114 @@
/* Neutral stacked strip above the input (queue rows are informational, not a warn state). */
/* Figma .FileContainerText 1:791: 776px wrapper around the inset 752px panel. */
.dock {
margin: 6px 0;
padding: 8px 12px;
border: 1px solid var(--dsw-alias-separator-primary);
border-radius: 10px;
background: var(--dsw-alias-bg-base);
box-sizing: border-box;
flex: none;
width: 100%;
max-width: 776px;
/* Eat InputBar's 6px top padding and tuck the panel 2px under the card;
the later composer sibling paints its surface and shadow over this edge. */
margin: 0 auto -10px;
padding: 2px 12px;
}
.title {
font-size: 12px;
font-weight: 500;
color: var(--dsw-alias-label-secondary);
.panel {
position: relative;
overflow: hidden;
width: 100%;
padding-top: 2px;
border-radius: 14px 14px 0 0;
background: var(--dsw-specific-tip);
}
.panel::after {
position: absolute;
inset: 0;
border: 1px solid var(--dsw-alias-border-l1);
border-bottom: none;
border-radius: inherit;
content: '';
pointer-events: none;
}
.list {
margin: 4px 0 0;
margin: 0;
padding: 0;
list-style: none;
}
.row {
overflow: hidden;
font-size: 12px;
line-height: 20px;
color: var(--dsw-alias-label-primary);
white-space: nowrap;
text-overflow: ellipsis;
box-sizing: border-box;
display: flex;
align-items: center;
gap: 10px;
width: 100%;
height: 36px;
padding: 4px 5px 4px 12px;
border-radius: 8px;
}
.preview,
.editor {
flex: 1 1 auto;
min-width: 0;
font: var(--dsw-font-xs-13);
font-family: Inter, var(--dsw-font-family);
}
.preview {
overflow: hidden;
color: var(--dsw-alias-label-primary-dimmed);
text-overflow: ellipsis;
white-space: nowrap;
word-break: break-word;
}
.editor {
box-sizing: border-box;
height: 28px;
padding: 0 8px;
border: 1px solid var(--dsw-alias-border-l2);
border-radius: 6px;
outline: none;
background: var(--dsw-alias-bg-base);
color: var(--dsw-alias-label-primary);
}
.editor:focus {
border-color: var(--dsw-alias-state-business-primary);
}
.actions {
display: flex;
flex: none;
align-items: center;
gap: 10px;
}
.action {
display: grid;
flex: none;
place-items: center;
width: 28px;
height: 28px;
padding: 0;
border: none;
border-radius: 999px;
background: transparent;
color: var(--dsw-alias-label-tertiary);
cursor: pointer;
}
.action:hover:not(:disabled) {
background: var(--dsw-alias-interactive-bg-hover);
}
.action:focus-visible {
outline: 2px solid var(--dsw-alias-label-tertiary);
outline-offset: -2px;
}
.action:disabled {
cursor: default;
opacity: 0.45;
}

View File

@@ -1,48 +1,185 @@
// Read-only queue dock entry (design v4 queue cut 1): renders the session's
// inbox mirror (session/queued frames + connect baseline) as one stacked
// strip above the input. No per-row actions — the host inbox has no
// addressable entries yet (queue cut 2 ledger).
// Queue dock entry: renders the authoritative transient inbox snapshot and
// addresses per-row mutations through the session-scoped conversation face.
//
// The 'conversation.input.dock' SlotMap declaration lives in
// ../contract/slots.ts beside the other input-region slots.
import type { Context } from 'cordis'
import { useEffect, useState } from 'react'
import type { PropsRuntime } from '@deepseek-ai/dsh-client-ui-slots'
import type {} from '@deepseek-ai/dsh-client-runtime/client'
import type { SessionId } from '@deepseek-ai/dsh-client-runtime/client'
import {
IconCheckOutline16, IconCloseOutline16, IconEditOutline16, IconTrashOutline16,
} from '@deepseek-ai/dsh-client-ui-primitives'
import type { QueueAction, QueueItemId } from '../contract/queue.ts'
import css from './QueueDock.module.css'
/** Queue operations injected by the session-scoped registration. */
export interface QueueDockInjected {
updateQueue: (itemId: QueueItemId, action: QueueAction) => Promise<void>
notify: (level: 'info' | 'error', text: string) => void
}
/** Full props of a dock entry: InputZone owner share + session standard kit + global seat. */
export type QueueDockProps = PropsRuntime<'conversation.input.dock'>
export type QueueDockProps = PropsRuntime<'conversation.input.dock'> & QueueDockInjected
/** Queue strip: one preview line per queued message; renders null when the queue is empty. */
export function QueueDock({ useSession }: QueueDockProps) {
export function QueueDock({ useSession, updateQueue, notify }: QueueDockProps) {
const queue = useSession(s => s.queue)
const [editing, setEditing] = useState<{ id: QueueItemId; text: string } | null>(null)
const [busy, setBusy] = useState<QueueItemId | null>(null)
useEffect(() => {
if (editing !== null && !queue.some(row => row.id === editing.id)) setEditing(null)
}, [editing, queue])
if (queue.length === 0) return null
const applyAction = async (
itemId: QueueItemId,
action: QueueAction,
failure: string,
): Promise<boolean> => {
setBusy(itemId)
try {
await updateQueue(itemId, action)
return true
} catch {
notify('error', failure)
return false
} finally {
setBusy(current => current === itemId ? null : current)
}
}
const saveEdit = async (): Promise<void> => {
if (editing === null || editing.text.trim() === '') return
if (await applyAction(
editing.id,
{ kind: 'edit', content: [{ type: 'text', text: editing.text }] },
'编辑失败:这条消息可能已经开始发送。',
)) setEditing(null)
}
return (
<div className={css.dock}>
<div className={css.title}> {queue.length} </div>
<ul className={css.list}>
{queue.map(row => (
<li key={row.key} className={css.row}>{row.preview}</li>
))}
</ul>
<div className={css.panel}>
<ul className={css.list}>
{queue.map(row => (
<li key={row.id} className={css.row}>
{editing?.id === row.id
? (
<input
autoFocus
className={css.editor}
aria-label="编辑排队消息"
value={editing.text}
onChange={(event) => { setEditing({ id: row.id, text: event.currentTarget.value }) }}
onKeyDown={(event) => {
if (event.key === 'Escape') {
setEditing(null)
return
}
if (event.key === 'Enter' && !event.nativeEvent.isComposing) {
event.preventDefault()
void saveEdit()
}
}}
/>
)
: <span className={css.preview}>{row.preview}</span>}
<div className={css.actions}>
{editing?.id === row.id
? (
<>
<button
type="button"
className={css.action}
aria-label="保存排队消息"
title="保存排队消息"
disabled={busy !== null || editing.text.trim() === ''}
onClick={() => { void saveEdit() }}
>
<IconCheckOutline16 size={14} />
</button>
<button
type="button"
className={css.action}
aria-label="取消编辑"
title="取消编辑"
disabled={busy !== null}
onClick={() => { setEditing(null) }}
>
<IconCloseOutline16 size={14} />
</button>
</>
)
: (
<>
<button
type="button"
className={css.action}
aria-label="编辑排队消息"
title={row.text === null ? '包含非文本内容,暂不支持编辑' : '编辑排队消息'}
disabled={busy !== null || row.text === null}
onClick={() => {
if (row.text !== null) setEditing({ id: row.id, text: row.text })
}}
>
<IconEditOutline16 size={14} />
</button>
<button
type="button"
className={css.action}
aria-label="删除排队消息"
title="删除排队消息"
disabled={busy !== null}
onClick={() => {
void applyAction(
row.id,
{ kind: 'remove' },
'删除失败:这条消息可能已经开始发送。',
)
}}
>
<IconTrashOutline16 size={14} />
</button>
</>
)}
</div>
</li>
))}
</ul>
</div>
</div>
)
}
/**
* The dock entry as a plain registrant plugin (bash posture).
* `inject: ['conversation']` is the ordering seam: the conversation service
* mounts after ui-conversation's slot registrations, so the
* 'conversation.input.dock' declaration is on the ledger by then.
* The dock entry as a plain registrant plugin. The conversation service is the
* ordering and action seam; session scopes provide the exact queue owner.
*/
export const queueDockEntry = {
name: 'conversation-queue-dock',
inject: ['slots', 'conversation'],
inject: ['slots', 'conversation', 'sessions'],
/**
* Register the queue strip into the input dock (list entry, order 0).
* @param ctx - registrant context (disposal rides ctx.effect inside slots.register).
*/
apply(ctx: Context): void {
ctx.slots.register({ name: 'conversation.input.dock', id: 'queue', order: 0 }, QueueDock)
ctx.slots.register({
name: 'conversation.input.dock',
id: 'queue',
order: 0,
inject: (sessionId: SessionId): QueueDockInjected => {
const actx = ctx.sessions.scope(sessionId)
if (actx === undefined) throw new Error(`queue dock: session "${sessionId}" resolved no scope`)
const conversation = actx.get('conversation')
if (conversation === undefined) throw new Error('queue dock: conversation service unavailable')
return {
updateQueue: (itemId, action) => conversation.updateQueue(itemId, action),
notify: (level, text) => { conversation.input.for(actx).notify(level, text) },
}
},
}, QueueDock)
},
}

View File

@@ -11,8 +11,8 @@ import type { QueuedMessage } from '../input/contract.ts'
/**
* Project a session's queue rows as a bare observable (subscribe/getSnapshot).
* The wiring layer (T5) overlays this onto InputState.queue; the runtime
* QueuedMessage and the input-contract QueuedMessage are structurally the
* same frozen shape ({key, preview}).
* QueuedMessage and the input-contract QueuedMessage are structurally
* identical.
* @param session - the resident session face.
* @returns the queue read face (snapshot reference stable while the queue is unchanged).
*/

View File

@@ -13,6 +13,7 @@ import type { Context } from 'cordis'
// error, so scope resolution goes through the sessions service (scopeOf
// method) instead of the standalone helper.
import type { ISessions, SessionFace, SessionId } from '@deepseek-ai/dsh-client-runtime/client'
import type { QueueAction, QueueItemId } from './contract/queue.ts'
import type { InputService } from './input/contract.ts'
/**
@@ -30,6 +31,13 @@ export interface IConversation {
* @returns completion; business failures reject (and land in promptError).
*/
send(text: string, mode: 'queue' | 'steer'): Promise<void>
/**
* Apply one operation to a pending queue occurrence.
* @param itemId - agent-owned inbox occurrence identity.
* @param action - edit or remove operation.
* @returns completion; business failures reject.
*/
updateQueue(itemId: QueueItemId, action: QueueAction): Promise<void>
/**
* Cancel the scoped session's in-flight turn.
* @returns completion; failures reject as in send.
@@ -71,6 +79,15 @@ export class ConversationService extends Service implements IConversation {
if (!result.ok) throw new Error(`conversation.send failed: ${result.error.code}: ${result.error.message}`)
}
/** Apply one operation to a pending queue occurrence. */
async updateQueue(itemId: QueueItemId, action: QueueAction): Promise<void> {
const session = this.scopedSession('updateQueue')
const result = await session.updateQueue(itemId, action)
if (!result.ok) {
throw new Error(`conversation.updateQueue failed: ${result.error.code}: ${result.error.message}`)
}
}
/** Cancel the scoped session's in-flight turn (failures land in promptError and reject, as in send). */
async cancel(): Promise<void> {
const session = this.scopedSession('cancel')

View File

@@ -1,20 +1,27 @@
// @vitest-environment jsdom
/**
* QueueDock rendering (web input-triggers queue cut 1): empty queue renders
* nothing, rows render one preview line each keyed by rpcId, and the strip
* follows queue changes through the useSession selector.
* QueueDock rendering and operations: authoritative rows, inline editing,
* removal, failure notices, and live retirement.
*/
import { afterEach, describe, expect, it } from 'vitest'
import { act, cleanup, render } from '@testing-library/react'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { act, cleanup, fireEvent, render, waitFor } from '@testing-library/react'
import { useSyncExternalStore } from 'react'
import type { ConversationSnapshot, QueuedMessage, SessionId, SessionListState } from '@deepseek-ai/dsh-client-runtime/client'
import type {
ConversationSnapshot, QueuedMessage, SessionId, SessionListState,
} from '@deepseek-ai/dsh-client-runtime/client'
import type { SnapshotSelectorHook } from '@deepseek-ai/dsh-client-ui-slots'
import type { QueueItemId } from '../src/client/contract/queue.ts'
import type { InputState } from '../src/client/input/contract.ts'
import { QueueDock, queueDockEntry } from '../src/client/queue/QueueDock.tsx'
import { QueueDock, queueDockEntry, type QueueDockInjected } from '../src/client/queue/QueueDock.tsx'
afterEach(cleanup)
const SID = 's1' as SessionId
const iid = (id: string): QueueItemId => id as QueueItemId
function row(id: string, text: string | null, preview = text ?? '[image]'): QueuedMessage {
return { id: iid(id), preview, text }
}
function snapshotWith(queue: QueuedMessage[]): ConversationSnapshot {
return {
@@ -24,31 +31,30 @@ function snapshotWith(queue: QueuedMessage[]): ConversationSnapshot {
}
}
/** Minimal live source backing the useSession stub (queue swaps notify subscribers). */
/** Minimal live source backing the useSession stub. */
function liveSession(initial: ConversationSnapshot) {
let snapshot = initial
const listeners = new Set<() => void>()
const useSession: SnapshotSelectorHook<ConversationSnapshot> = sel =>
const useSession: SnapshotSelectorHook<ConversationSnapshot> = selector =>
useSyncExternalStore(
(fn) => {
listeners.add(fn)
return () => listeners.delete(fn)
(listener) => {
listeners.add(listener)
return () => listeners.delete(listener)
},
() => sel(snapshot),
() => selector(snapshot),
)
return {
useSession,
push(next: ConversationSnapshot): void {
snapshot = next
for (const fn of [...listeners]) fn()
for (const listener of [...listeners]) listener()
},
}
}
/** InputZone owner stub (the dock reads useSession only; the zone fields satisfy the owner share). */
const INPUT_STATE: InputState = { draft: '', draftRev: 0, phase: 'plain', occurrences: [], queue: [] }
function kitFor(snapshot: ConversationSnapshot) {
function kitFor(snapshot: ConversationSnapshot, injected: Partial<QueueDockInjected> = {}) {
return {
sessionId: SID,
useSessions: (() => { throw new Error('unused') }) as unknown as SnapshotSelectorHook<SessionListState>,
@@ -58,6 +64,9 @@ function kitFor(snapshot: ConversationSnapshot) {
inputActions: { setDraft: () => {}, submit: () => {} } as never,
session: snapshot,
input: INPUT_STATE,
updateQueue: vi.fn(() => Promise.resolve()),
notify: vi.fn(),
...injected,
}
}
@@ -69,20 +78,118 @@ describe('QueueDock', () => {
expect(container.innerHTML).toBe('')
})
it('renders one preview row per queued message with the count strip', () => {
it('renders active actions and disables editing for mixed-content rows', () => {
const snap = snapshotWith([
{ key: 'p-1', preview: '第一条排队消息' },
{ key: 'p-2', preview: 'second queued line' },
row('i-1', '第一条排队消息'),
row('i-2', null, 'image [image]'),
])
const source = liveSession(snap)
const { container } = render(<QueueDock {...kitFor(snap)} useSession={source.useSession} />)
expect(container.textContent).toContain('已排队 2 条')
const rows = [...container.querySelectorAll('li')]
expect(rows.map(r => r.textContent)).toEqual(['第一条排队消息', 'second queued line'])
expect([...container.querySelectorAll('li')].map(item => item.textContent))
.toEqual(['第一条排队消息', 'image [image]'])
expect(container.querySelectorAll('button')).toHaveLength(4)
expect(container.querySelectorAll('[aria-label="编辑排队消息"]')).toHaveLength(2)
expect(container.querySelectorAll('[aria-label="删除排队消息"]')).toHaveLength(2)
expect(container.querySelectorAll('[aria-label="立即发送排队消息"]')).toHaveLength(0)
expect((container.querySelectorAll('[aria-label="编辑排队消息"]')[0] as HTMLButtonElement).disabled).toBe(false)
expect((container.querySelectorAll('[aria-label="编辑排队消息"]')[1] as HTMLButtonElement).disabled).toBe(true)
expect(container.querySelectorAll('[aria-label="编辑排队消息"]')[1]?.getAttribute('title'))
.toBe('包含非文本内容,暂不支持编辑')
})
it('follows queue changes: retirement empties the strip back to null', () => {
const snap = snapshotWith([{ key: 'p-1', preview: '在场' }])
it('edits text inline with save and cancel controls, then saves with the same item identity', async () => {
const snap = snapshotWith([row('i-edit', 'before')])
const source = liveSession(snap)
const updateQueue = vi.fn(() => Promise.resolve())
const { getByLabelText, queryByLabelText } = render(
<QueueDock {...kitFor(snap, { updateQueue })} useSession={source.useSession} />,
)
fireEvent.click(getByLabelText('编辑排队消息'))
const editor = getByLabelText('编辑排队消息') as HTMLInputElement
expect(getByLabelText('保存排队消息')).toBeTruthy()
expect(getByLabelText('取消编辑')).toBeTruthy()
expect(queryByLabelText('删除排队消息')).toBeNull()
fireEvent.change(editor, { target: { value: 'after' } })
fireEvent.keyDown(editor, { key: 'Enter' })
await waitFor(() => {
expect(updateQueue).toHaveBeenCalledWith(iid('i-edit'), {
kind: 'edit',
content: [{ type: 'text', text: 'after' }],
})
})
})
it('cancels an edit by button or Escape without mutating the queue', () => {
const snap = snapshotWith([row('i-edit', 'before')])
const source = liveSession(snap)
const updateQueue = vi.fn(() => Promise.resolve())
const { getByLabelText, getByText } = render(
<QueueDock {...kitFor(snap, { updateQueue })} useSession={source.useSession} />,
)
fireEvent.click(getByLabelText('编辑排队消息'))
fireEvent.change(getByLabelText('编辑排队消息'), { target: { value: 'abandoned' } })
fireEvent.click(getByLabelText('取消编辑'))
expect(getByText('before')).toBeTruthy()
fireEvent.click(getByLabelText('编辑排队消息'))
fireEvent.keyDown(getByLabelText('编辑排队消息'), { key: 'Escape' })
expect(getByText('before')).toBeTruthy()
expect(updateQueue).not.toHaveBeenCalled()
})
it('keeps editing during IME composition and disables a blank save', () => {
const snap = snapshotWith([row('i-edit', 'before')])
const source = liveSession(snap)
const updateQueue = vi.fn(() => Promise.resolve())
const { getByLabelText } = render(
<QueueDock {...kitFor(snap, { updateQueue })} useSession={source.useSession} />,
)
fireEvent.click(getByLabelText('编辑排队消息'))
const editor = getByLabelText('编辑排队消息')
fireEvent.change(editor, { target: { value: ' ' } })
expect(getByLabelText('保存排队消息')).toHaveProperty('disabled', true)
fireEvent.change(editor, { target: { value: '输入中' } })
fireEvent.keyDown(editor, { key: 'Enter', isComposing: true })
expect(updateQueue).not.toHaveBeenCalled()
expect(getByLabelText('编辑排队消息')).toBeTruthy()
})
it('removes the addressed row', async () => {
const snap = snapshotWith([row('i-1', 'one'), row('i-2', 'two')])
const source = liveSession(snap)
const updateQueue = vi.fn(() => Promise.resolve())
const { getAllByLabelText } = render(
<QueueDock {...kitFor(snap, { updateQueue })} useSession={source.useSession} />,
)
fireEvent.click(getAllByLabelText('删除排队消息')[0]!)
await waitFor(() => {
expect(updateQueue).toHaveBeenCalledWith(iid('i-1'), { kind: 'remove' })
})
})
it('keeps the row and surfaces a notice when an operation loses the claim race', async () => {
const snap = snapshotWith([row('i-race', 'pending')])
const source = liveSession(snap)
const notify = vi.fn()
const updateQueue = vi.fn(() => Promise.reject(new Error('not found')))
const { getByLabelText, getByText } = render(
<QueueDock {...kitFor(snap, { updateQueue, notify })} useSession={source.useSession} />,
)
fireEvent.click(getByLabelText('删除排队消息'))
await waitFor(() => {
expect(notify).toHaveBeenCalledWith('error', '删除失败:这条消息可能已经开始发送。')
})
expect(getByText('pending')).toBeTruthy()
})
it('follows authoritative retirement back to null', () => {
const snap = snapshotWith([row('i-1', '在场')])
const source = liveSession(snap)
const { container } = render(<QueueDock {...kitFor(snap)} useSession={source.useSession} />)
expect(container.textContent).toContain('在场')
@@ -90,11 +197,9 @@ describe('QueueDock', () => {
expect(container.innerHTML).toBe('')
})
it('ships the registrant plugin shape (list entry into conversation.input.dock)', () => {
// Registration itself runs under T5's slot declaration; here we pin the
// frozen registration surface so the wiring layer can mount it verbatim.
it('ships the session-scoped registrant plugin shape', () => {
expect(queueDockEntry.name).toBe('conversation-queue-dock')
expect(queueDockEntry.inject).toEqual(['slots', 'conversation'])
expect(queueDockEntry.inject).toEqual(['slots', 'conversation', 'sessions'])
expect(typeof queueDockEntry.apply).toBe('function')
})
})

View File

@@ -12,11 +12,12 @@ import { InputHub } from '../src/client/input/hub.ts'
async function bench() {
const runtime = await SlotTestRuntime.create()
const prompt = vi.fn(() => Promise.resolve({ ok: true as const, value: { accepted: true as const } }))
const updateQueue = vi.fn(() => Promise.resolve({ ok: true as const, value: { accepted: true as const } }))
const cancel = vi.fn(() => Promise.resolve({ ok: true as const, value: { accepted: true as const } }))
const loadOlder = vi.fn(() => Promise.resolve())
await runtime.sessions.add({
id: 's1',
session: { prompt, cancel, loadOlder },
session: { prompt, updateQueue, cancel, loadOlder },
})
// config.input is required (the apply shares its hub with the inject
// factories); the bench passes its own instance explicitly.
@@ -26,16 +27,18 @@ async function bench() {
await fiber.await()
const root = runtime.ctx.get('conversation') as ConversationService
const scoped = runtime.sessions.scope('s1')!.get('conversation') as ConversationService
return { runtime, root, scoped, prompt, cancel, loadOlder }
return { runtime, root, scoped, prompt, updateQueue, cancel, loadOlder }
}
describe('ConversationService', () => {
it('routes operations through the public Session binding', async () => {
const b = await bench()
await b.scoped.send('hello', 'steer')
await b.scoped.updateQueue('item-1' as never, { kind: 'remove' })
await b.scoped.cancel()
await b.scoped.loadOlder()
expect(b.prompt).toHaveBeenCalledWith([{ type: 'text', text: 'hello' }], 'steer')
expect(b.updateQueue).toHaveBeenCalledWith('item-1', { kind: 'remove' })
expect(b.cancel).toHaveBeenCalledOnce()
expect(b.loadOlder).toHaveBeenCalledOnce()
await b.runtime.dispose()