fix(chat): retain assistant nodes and steering echoes
This commit is contained in:
1 parent
92046a8abc
commit
3b4ecc9d20
14 files changed
+316
-59
No files matched your search
@@ -50,6 +50,7 @@ const AUTH_PROVIDER_MESSAGE = 'Authentication Fails, Your api key: sk-preview-se
|
||||
// model call.
|
||||
const PROMPT = 'Reply with a one-sentence description of event sourcing, then stop.'
|
||||
const RUNNING_DRAFT = 'Queue this follow-up while the current turn is running.'
|
||||
const RETRY_PARTIAL = 'This partial reply must disappear when the attempt fails.'
|
||||
|
||||
/** turn/end reasons observed, in order. */
|
||||
function turnEndReasons(events: SessionEvent[]): string[] {
|
||||
@@ -270,21 +271,44 @@ describe('web e2e: live-turn interactions (cancel / error / retry)', () => {
|
||||
expect(derived).toHaveLength(1)
|
||||
await launch(() => ({
|
||||
patches: [
|
||||
{ at: 0, entry: { kind: 'throw', chunks: [], message: 'upstream 503', code: 'SERVER' } },
|
||||
{ at: 0, entry: {
|
||||
kind: 'throw', message: 'upstream 503', code: 'SERVER',
|
||||
chunks: [
|
||||
{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
{ type: 'text-delta', index: 0, text: RETRY_PARTIAL },
|
||||
],
|
||||
} },
|
||||
// Append the fixture's own success as the retry attempt — single-
|
||||
// sourced from the recording, never copied into a committed sidecar.
|
||||
{ at: 1, entry: derived[0]! },
|
||||
],
|
||||
}))
|
||||
onTestFailed(() => saveFailureShot(page, 'web-e2e-retry'))
|
||||
// llm-retry backs off ~500ms before the second attempt.
|
||||
const { settled } = await sendPrompt(60_000)
|
||||
await settled
|
||||
const releaseFailure = Promise.withResolvers<undefined>()
|
||||
// Retirement must follow a visible partial, not race the first browser paint.
|
||||
const stopHolding = scaffold!.ctx.on('llm/stream', async function* (_request, next) {
|
||||
for await (const chunk of next()) {
|
||||
yield chunk
|
||||
if (chunk.type === 'text-delta' && chunk.text === RETRY_PARTIAL) await releaseFailure.promise
|
||||
}
|
||||
})
|
||||
try {
|
||||
const { settled } = await sendPrompt(60_000)
|
||||
await page.getByText(RETRY_PARTIAL, { exact: true }).waitFor({ timeout: 15_000 })
|
||||
releaseFailure.resolve(undefined)
|
||||
await settled
|
||||
} finally {
|
||||
releaseFailure.resolve(undefined)
|
||||
stopHolding()
|
||||
}
|
||||
expect(turnEndReasons(sessionEvents).at(-1)).toBe('completed')
|
||||
// The durable retry record proves the second attempt (request/header logs
|
||||
// only on change, so attempt count is invisible there).
|
||||
expect(sessionEvents.filter(e => e.type === 'llm/retry').length).toBeGreaterThanOrEqual(1)
|
||||
await expect.poll(() => page.getByText('event sourcing', { exact: false }).count(), { timeout: 10_000 }).toBeGreaterThan(0)
|
||||
await expect.poll(() => page.getByText(RETRY_PARTIAL, { exact: true }).count()).toBe(0)
|
||||
expect(tripwire.pageErrors).toEqual([])
|
||||
expect(tripwire.warnings).toEqual([])
|
||||
// Golden of the recovered end-state: the discarded partial stays absent,
|
||||
// while the settled retry row remains as durable recovery context.
|
||||
const snapshot = await captureStableAria(page, '[class*="centerCol"]', scaffold!.workspaceCwd)
|
||||
|
||||
@@ -235,9 +235,28 @@ describe('web e2e: composer shortcut steers directly', () => {
|
||||
await composer.waitFor({ timeout: 30_000 })
|
||||
const pendingSteering = page.locator('[data-pending-steering]').filter({ hasText: STEER })
|
||||
await pendingSteering.waitFor({ timeout: 10_000 })
|
||||
await composer.getByRole('radio', { name: 'Yes' }).click()
|
||||
await composer.getByRole('radio', { name: 'Yes' }).press('Enter')
|
||||
await settled
|
||||
const admission = Promise.withResolvers<undefined>()
|
||||
let claimed = false
|
||||
const stopHolding = scaffold.ctx.on('agent/pre-step', async (payload, next) => {
|
||||
if (payload.messages.some(message => message.content.some(block => block.type === 'text' && block.text === STEER))) {
|
||||
claimed = true
|
||||
await admission.promise
|
||||
}
|
||||
return next()
|
||||
})
|
||||
try {
|
||||
await composer.getByRole('radio', { name: 'Yes' }).click()
|
||||
await composer.getByRole('radio', { name: 'Yes' }).press('Enter')
|
||||
await expect.poll(() => claimed, { timeout: 10_000 }).toBe(true)
|
||||
await page.evaluate(() => new Promise<void>((resolve) => {
|
||||
requestAnimationFrame(() => requestAnimationFrame(() => { resolve() }))
|
||||
}))
|
||||
expect(await page.getByText(STEER, { exact: true }).count()).toBe(1)
|
||||
} finally {
|
||||
admission.resolve(undefined)
|
||||
stopHolding()
|
||||
await settled
|
||||
}
|
||||
|
||||
const steerEvents = claimedMessages(sessionEvents, STEER)
|
||||
expect(steerEvents).toHaveLength(1)
|
||||
|
||||
@@ -6,6 +6,7 @@ import { randomUUID } from '@deepseek-ai/dsh-util-crypto'
|
||||
import type { AttachmentIdType, FileAttachmentRef, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
|
||||
import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
|
||||
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
|
||||
import type { UserMessage } from '@deepseek-ai/dsh-llm/types'
|
||||
import { SessionLogOffset, SessionSeq, type SessionId } from '@deepseek-ai/dsh-session/types'
|
||||
import { SessionEventStream } from '../transport.ts'
|
||||
import type { SessionJournalChange } from '../transport.ts'
|
||||
@@ -25,7 +26,7 @@ import type {
|
||||
} from '../contract/snapshot.ts'
|
||||
import { MutableSessionEventSource } from '../contract/events.ts'
|
||||
import type {
|
||||
SessionEventLikeEntry, SessionLiveEventEntry,
|
||||
SessionEventLike, SessionEventLikeEntry, SessionLiveEventEntry,
|
||||
} from '../contract/events.ts'
|
||||
import { Notifier } from './notifier.ts'
|
||||
import { isRemoteFailure } from '@deepseek-ai/dsh-api-gateway/client'
|
||||
@@ -124,6 +125,9 @@ export class Session implements SessionFace {
|
||||
/** Per-echo settlement state; `retiring` latches the first observation so a
|
||||
* Inbox projection and its durable event cannot both retire one echo. */
|
||||
private readonly submissionSettlements = new Map<SessionRequestId, {
|
||||
readonly placement: PendingSubmission['placement']
|
||||
/** Latest received next-step position; null means claimed but not yet admitted. */
|
||||
receipt?: { readonly seq: number; readonly index: number | null }
|
||||
readonly onRetire?: ((retirement: PendingSubmissionRetirement) => void) | undefined
|
||||
retiring: boolean
|
||||
}>()
|
||||
@@ -209,16 +213,15 @@ export class Session implements SessionFace {
|
||||
*/
|
||||
beginSubmission(input: BeginSubmissionInput): SubmissionHandle {
|
||||
const requestId = randomUUID() as SessionRequestId
|
||||
const placement = this.running ? input.mode === 'steer' ? 'steering' : 'queued' : 'transcript'
|
||||
this.pendingSubmissions = [...this.pendingSubmissions, {
|
||||
requestId,
|
||||
placement: this.running
|
||||
? input.mode === 'steer' ? 'steering' : 'queued'
|
||||
: 'transcript',
|
||||
placement,
|
||||
time: Date.now(),
|
||||
text: input.text,
|
||||
attachments: input.attachments,
|
||||
}]
|
||||
this.submissionSettlements.set(requestId, { onRetire: input.onRetire, retiring: false })
|
||||
this.submissionSettlements.set(requestId, { placement, onRetire: input.onRetire, retiring: false })
|
||||
// The blank → engaging edge flips here, ahead of prompt(): the composer
|
||||
// docks and the echo renders on the click's own frame.
|
||||
this.promptAttempted = true
|
||||
@@ -666,6 +669,10 @@ export class Session implements SessionFace {
|
||||
if (projections !== undefined) this.projections.seed(projections)
|
||||
this.eventSource.replace(visible, hasMore)
|
||||
for (const entry of visible) this.observeSubmissionEvent(entry.event)
|
||||
if (projections !== undefined) {
|
||||
const inbox = projections.values.inbox as InboxState | undefined
|
||||
this.observeSteeringInsertions(inbox?.['next-step'] ?? [], 0, projections.asOfSeq)
|
||||
}
|
||||
this.notifier.markDirty()
|
||||
}
|
||||
|
||||
@@ -724,36 +731,58 @@ export class Session implements SessionFace {
|
||||
}
|
||||
|
||||
/** Observe durable acceptance even when insertion and claim share one projection notification. */
|
||||
private observeSubmissionEvent(event: { readonly type: string; readonly data?: unknown }): void {
|
||||
private observeSubmissionEvent(event: SessionEventLike): void {
|
||||
if (this.submissionSettlements.size === 0) return
|
||||
if (event.type === 'agent/inbox/spliced') {
|
||||
const splice = event.data as { readonly inserted?: unknown } | undefined
|
||||
if (Array.isArray(splice?.inserted)) {
|
||||
for (const message of splice.inserted) this.observeSubmissionEvent({ type: 'user/message', data: message })
|
||||
const { target, start, removedCount = 0, inserted, outcome } = event.data
|
||||
if (target === 'next-step') {
|
||||
for (const [requestId, settlement] of this.submissionSettlements) {
|
||||
const receipt = settlement.receipt
|
||||
if (receipt === undefined || receipt.index === null || receipt.seq >= event.seq) continue
|
||||
const removed = receipt.index >= start && receipt.index < start + removedCount
|
||||
if (removed && outcome === 'canceled') this.retireFailedSubmission(requestId)
|
||||
else settlement.receipt = {
|
||||
seq: event.seq,
|
||||
index: removed ? null : receipt.index < start ? receipt.index : receipt.index + inserted.length - removedCount,
|
||||
}
|
||||
}
|
||||
this.observeSteeringInsertions(inserted, start, event.seq)
|
||||
}
|
||||
for (const message of inserted) this.observeSubmissionMessage(message, false)
|
||||
return
|
||||
}
|
||||
if (event.type === 'request/context' || event.type === 'turn/end') {
|
||||
for (const [requestId, settlement] of this.submissionSettlements) {
|
||||
if (settlement.receipt?.index === null && settlement.receipt.seq < event.seq) this.retireFailedSubmission(requestId)
|
||||
}
|
||||
return
|
||||
}
|
||||
if (event.type !== 'user/message') return
|
||||
// Structural read: window entries may be compact history records, so the
|
||||
// fields are narrowed rather than trusted (same posture as Conversation
|
||||
// assembly matchers).
|
||||
const data = event.data as { readonly source?: unknown; readonly content?: unknown } | undefined
|
||||
const source = data?.source as { readonly kind?: unknown; readonly rpcId?: unknown } | undefined
|
||||
if (source?.kind !== 'user' || typeof source.rpcId !== 'string') return
|
||||
this.scheduleObservedRetirement(source.rpcId as SessionRequestId, attachmentRefsIn(data?.content))
|
||||
if (event.type === 'user/message') this.observeSubmissionMessage(event.data, true)
|
||||
}
|
||||
|
||||
/** Retire local echoes when their accepted messages appear in the durable Inbox projection. */
|
||||
private observeSteeringInsertions(messages: readonly UserMessage[], start: number, seq: number): void {
|
||||
for (const [index, message] of messages.entries()) {
|
||||
const source = message.source
|
||||
if (source.kind !== 'user' || !('rpcId' in source)) continue
|
||||
const settlement = this.submissionSettlements.get(source.rpcId)
|
||||
if (settlement?.placement !== 'steering' || settlement.retiring || (settlement.receipt?.seq ?? -1) > seq) continue
|
||||
settlement.receipt = { seq, index: start + index }
|
||||
}
|
||||
}
|
||||
|
||||
private observeSubmissionMessage(message: UserMessage, admitted: boolean): void {
|
||||
const source = message.source
|
||||
if (source.kind !== 'user' || !('rpcId' in source)) return
|
||||
if (!admitted && this.submissionSettlements.get(source.rpcId)?.placement === 'steering') return
|
||||
this.scheduleObservedRetirement(source.rpcId, attachmentRefsIn(message.content))
|
||||
}
|
||||
|
||||
/** Retire non-steering echoes when the Inbox accepts their queue occurrences. */
|
||||
private observeSubmissionInbox(): void {
|
||||
if (this.submissionSettlements.size === 0) return
|
||||
const inbox = this.projections.get('inbox') as InboxState | undefined
|
||||
if (inbox === undefined) return
|
||||
for (const message of [...inbox['next-turn'], ...inbox['next-step']]) {
|
||||
const source = message.source
|
||||
if (source.kind === 'user' && 'rpcId' in source) {
|
||||
this.scheduleObservedRetirement(source.rpcId, attachmentRefsIn(message.content))
|
||||
}
|
||||
}
|
||||
for (const message of [...inbox['next-turn'], ...inbox['next-step']]) this.observeSubmissionMessage(message, false)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -193,6 +193,77 @@ describe('observed retirement', () => {
|
||||
expect(onRetire).toHaveBeenCalledExactlyOnceWith({ reason: 'observed', attachments: refs })
|
||||
})
|
||||
|
||||
it('keeps a running steer echo through accepted and claimed Inbox states until transcript admission', async ({ mock, start }) => {
|
||||
const session = await sessionBench(mock, start, SID)
|
||||
await session.open()
|
||||
session.handleRunning(true)
|
||||
const onRetire = vi.fn()
|
||||
const handle = session.beginSubmission({ mode: 'steer', text: 'accepted', attachments: [], onRetire })
|
||||
const refs = [imageRef('steer-image')]
|
||||
const message = queuedItem(handle.requestId, refs)
|
||||
const echo = session.getSnapshot().pendingSubmissions[0]
|
||||
session.projections.apply('inbox', { 'next-turn': [], 'next-step': [message] }, SessionSeq(0))
|
||||
await settleFrames()
|
||||
expect(session.getSnapshot().pendingSubmissions).toEqual([echo])
|
||||
await pushEvent(mock, {
|
||||
type: 'agent/inbox/spliced', seq: SessionSeq(0), time: 1,
|
||||
data: { target: 'next-step', start: 0, inserted: [message] },
|
||||
})
|
||||
await pushEvent(mock, {
|
||||
type: 'agent/inbox/spliced', seq: SessionSeq(1), time: 2,
|
||||
data: { target: 'next-step', start: 0, removedCount: 1, inserted: [] },
|
||||
})
|
||||
session.projections.apply('inbox', { 'next-turn': [], 'next-step': [] }, SessionSeq(1))
|
||||
await settleFrames()
|
||||
expect(session.getSnapshot().pendingSubmissions).toEqual([echo])
|
||||
expect(onRetire).not.toHaveBeenCalled()
|
||||
await pushEvent(mock, {
|
||||
type: 'user/message', seq: SessionSeq(2), time: 3, surfaceOp: 'append', data: message,
|
||||
})
|
||||
await settleFrames()
|
||||
expect(session.getSnapshot().pendingSubmissions).toEqual([])
|
||||
expect(onRetire).toHaveBeenCalledExactlyOnceWith({ reason: 'observed', attachments: refs })
|
||||
})
|
||||
|
||||
it('retires canceled or unadmitted steering without removing still-pending input', async ({ mock, start }) => {
|
||||
const session = await sessionBench(mock, start, SID)
|
||||
await session.open()
|
||||
session.handleRunning(true)
|
||||
const canceled = vi.fn()
|
||||
const rejected = vi.fn()
|
||||
const first = session.beginSubmission({ mode: 'steer', text: 'cancel', attachments: [], onRetire: canceled })
|
||||
const second = session.beginSubmission({ mode: 'steer', text: 'reject', attachments: [], onRetire: rejected })
|
||||
const unclaimed = session.beginSubmission({ mode: 'steer', text: 'later', attachments: [] })
|
||||
const foreign = createUserMessage({ source: { kind: 'user' }, content: [] })
|
||||
await pushEvent(mock, {
|
||||
type: 'agent/inbox/spliced', seq: SessionSeq(0), time: 1,
|
||||
data: { target: 'next-step', start: 0, inserted: [queuedItem(first.requestId), queuedItem(second.requestId)] },
|
||||
})
|
||||
await pushEvent(mock, {
|
||||
type: 'agent/inbox/spliced', seq: SessionSeq(1), time: 2,
|
||||
data: { target: 'next-step', start: 0, inserted: [foreign] },
|
||||
})
|
||||
await pushEvent(mock, {
|
||||
type: 'agent/inbox/spliced', seq: SessionSeq(2), time: 3,
|
||||
data: { target: 'next-step', start: 1, removedCount: 1, inserted: [], outcome: 'canceled' },
|
||||
})
|
||||
expect(canceled).toHaveBeenCalledExactlyOnceWith({ reason: 'failed' })
|
||||
expect(rejected).not.toHaveBeenCalled()
|
||||
await pushEvent(mock, {
|
||||
type: 'agent/inbox/spliced', seq: SessionSeq(3), time: 4,
|
||||
data: { target: 'next-step', start: 0, removedCount: 2, inserted: [] },
|
||||
})
|
||||
await pushEvent(mock, {
|
||||
type: 'agent/inbox/spliced', seq: SessionSeq(4), time: 5,
|
||||
data: { target: 'next-step', start: 0, inserted: [queuedItem(unclaimed.requestId)] },
|
||||
})
|
||||
await pushEvent(mock, {
|
||||
type: 'turn/end', seq: SessionSeq(5), time: 6, data: { turn: 1, reason: { kind: 'blocked' } },
|
||||
})
|
||||
expect(rejected).toHaveBeenCalledExactlyOnceWith({ reason: 'failed' })
|
||||
expect(session.getSnapshot().pendingSubmissions.map(echo => echo.requestId)).toEqual([unclaimed.requestId])
|
||||
})
|
||||
|
||||
it('a queue occurrence carrying the rpcId retires the echo (running-turn submissions)', async ({ mock, start }) => {
|
||||
const session = await sessionBench(mock, start, SID)
|
||||
const retirements: PendingSubmissionRetirement[] = []
|
||||
|
||||
@@ -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-chat/README.md
|
||||
README.md: b383d74805dfd9e37c4b69136c15a63ca087e10d
|
||||
README.zh.md: 046036f0919598afb363a97a984d6a137550d12b
|
||||
README.md: 34ac2d6e79b6ec67583b7a0f85d48e71f7fb27b1
|
||||
README.zh.md: bcb521100ff11b59e1e1c2b5a47b45cabb1ce57d
|
||||
@@ -40,6 +40,8 @@ Settings → General → Open chat links in selects the destination for ordinary
|
||||
|
||||
Chat omits system-prompt, ordinary Context injection, and `permission` command rows in every work-details mode. The filter changes neither recorded Session events nor Trajectory inspection. Non-human Turn triggers remain independent notices; other command rows remain in Chat.
|
||||
|
||||
When an Assistant attempt retires without a visible message, Chat hides its already-published Node instead of removing its key. A retry in the same Step reuses that key when visible content returns. This also applies when the loaded window lacks the Step start.
|
||||
|
||||
<a id="command-and-failure-rows"></a>
|
||||
## Command and failure rows
|
||||
|
||||
@@ -70,6 +72,8 @@ The completed-turn action footer follows the recorded Turn end. Its action row s
|
||||
|
||||
Preference menus restore focus to their trigger without scrolling before publishing a new selection.
|
||||
|
||||
Local steering echoes remain mounted through Inbox acceptance and claim until the durable message arrives. Matching Host-pending rows stay hidden during this handoff, and the same request does not trigger tail following twice.
|
||||
|
||||
Work-details modes control process-group display and reasoning previews; eligible completed Turns fold their process without hiding the final answer. The [business-rule reference](src/client/conversation-nodes/README.md#display-modes) contains the mode table, title behavior, whole-Turn eligibility, clocks, and disclosure resets.
|
||||
|
||||
-----
|
||||
|
||||
@@ -40,6 +40,8 @@ Chat 在节点列表外通过一个 `MarkdownDelegateProvider` 提供文件及 H
|
||||
|
||||
Chat 在所有工作过程展示模式下都不显示系统提示词行、普通上下文注入和 `permission` 命令行。该过滤不改变已记录的 Session 事件或 Trajectory 查看能力。非人工轮次触发仍作为独立通知显示,其他命令行仍保留在 Chat 中。
|
||||
|
||||
Assistant 尝试结束且没有可见消息时,Chat 隐藏已发布的 Node,不移除其 key。同一 Step 的重试再次产生可见内容时,复用该 key。已加载窗口缺少 Step 起点时也遵循此规则。
|
||||
|
||||
<a id="command-and-failure-rows"></a>
|
||||
## 指令与失败行
|
||||
|
||||
@@ -70,6 +72,8 @@ Chat 在所有工作过程展示模式下都不显示系统提示词行、普通
|
||||
|
||||
偏好菜单在发布新选择之前,先将焦点还给触发按钮,且不引起滚动。
|
||||
|
||||
本地 steering 回显在 Inbox 接受与领取期间保持挂载,直到持久消息到达。交接期间隐藏同请求的 Host 待处理行,同一请求不会重复触发跟随底部。
|
||||
|
||||
工作过程展示模式控制过程组显示与推理预览;符合条件的已完成轮次折叠过程,但不隐藏最终答案。[业务规则明细](src/client/conversation-nodes/README.zh.md#display-modes) 统一说明模式表、组头行为、整轮折叠资格、时钟与开合重置。
|
||||
|
||||
-----
|
||||
|
||||
@@ -38,7 +38,7 @@ function observedRpcIds(
|
||||
order: readonly string[],
|
||||
nodes: ChatSnapshot['nodes'],
|
||||
inbox: InboxState | undefined,
|
||||
): ReadonlySet<string> {
|
||||
): { readonly durable: ReadonlySet<string>; readonly pending: ReadonlySet<string> } {
|
||||
const observed = new Set<string>()
|
||||
for (const key of order) {
|
||||
const node = nodes.get(key)
|
||||
@@ -48,10 +48,11 @@ function observedRpcIds(
|
||||
| undefined
|
||||
if (source?.kind === 'user' && typeof source.rpcId === 'string') observed.add(source.rpcId)
|
||||
}
|
||||
const pending = new Set<string>()
|
||||
for (const { source } of [...inbox?.['next-turn'] ?? [], ...inbox?.['next-step'] ?? []]) {
|
||||
if (source.kind === 'user' && 'rpcId' in source) observed.add(source.rpcId)
|
||||
if (source.kind === 'user' && 'rpcId' in source) pending.add(source.rpcId)
|
||||
}
|
||||
return observed
|
||||
return { durable: observed, pending }
|
||||
}
|
||||
|
||||
type ChatNodeListProps = Omit<ComponentProps<typeof ChatNodeSeat>, 'nodeKey' | 'groupPart'> & {
|
||||
@@ -141,7 +142,7 @@ export function ChatView({
|
||||
setFileOpenBusy(false)
|
||||
}, [])
|
||||
|
||||
const pendingSteering = useMemo(
|
||||
const inboxSteering = useMemo(
|
||||
() => inbox?.['next-step'].filter(message => message.source.kind === 'user') ?? [],
|
||||
[inbox],
|
||||
)
|
||||
@@ -153,9 +154,15 @@ export function ChatView({
|
||||
if (pendingSubmissions.length === 0) return pendingSubmissions
|
||||
const observed = observedRpcIds(order, nodeStore, inbox)
|
||||
return pendingSubmissions.filter(submission => (
|
||||
submission.placement !== 'queued' && !observed.has(submission.requestId)
|
||||
submission.placement !== 'queued' && !observed.durable.has(submission.requestId)
|
||||
&& (submission.placement === 'steering' || !observed.pending.has(submission.requestId))
|
||||
))
|
||||
}, [pendingSubmissions, order, nodeStore, inbox])
|
||||
const pendingSteering = useMemo(() => {
|
||||
const local = new Set(visibleSubmissions.filter(submission => submission.placement === 'steering')
|
||||
.map(submission => submission.requestId))
|
||||
return inboxSteering.filter(({ source }) => source.kind !== 'user' || !('rpcId' in source) || !local.has(source.rpcId))
|
||||
}, [inboxSteering, visibleSubmissions])
|
||||
const renderMessageImages = useCallback<RenderMessageImages>(
|
||||
owner => renderSlot('conversation.message.images', { ...owner, loadImage }),
|
||||
[loadImage, renderSlot],
|
||||
@@ -164,11 +171,14 @@ export function ChatView({
|
||||
const firstKey = order[0]
|
||||
const firstSeq = firstKey === undefined ? null : nodeStore.get(firstKey)?.anchorSeq ?? null
|
||||
const lastKey = order.at(-1) ?? null
|
||||
const latestSteering = pendingSteering.at(-1)
|
||||
const steeringId = latestSteering?.source.kind === 'user' && 'rpcId' in latestSteering.source
|
||||
? latestSteering.source.rpcId : latestSteering?.id ?? null
|
||||
const scroll = useChatScroll({
|
||||
ready: openState === 'open',
|
||||
order, firstSeq, lastKey, running, loadingOlder, hasMore, chatScroll, loadOlder, loadThrough,
|
||||
lastIsUser: lastKey !== null && nodeStore.get(lastKey)?.kind === 'user',
|
||||
steeringId: pendingSteering.at(-1)?.id ?? null,
|
||||
steeringId,
|
||||
submissionId: visibleSubmissions.at(-1)?.requestId ?? null,
|
||||
loadedTurns: turnNavigationItems,
|
||||
})
|
||||
|
||||
@@ -53,8 +53,10 @@ export function useChatScroll(input: ChatScrollInput): ChatScrollState {
|
||||
const current = content.current.input
|
||||
const previous = content.current.applied
|
||||
const ownInput = (current.lastIsUser && current.lastKey !== previous?.lastKey)
|
||||
|| (current.steeringId !== null && current.steeringId !== previous?.steeringId)
|
||||
|| (current.submissionId !== null && current.submissionId !== previous?.submissionId)
|
||||
|| (current.steeringId !== null && current.steeringId !== previous?.steeringId
|
||||
&& current.steeringId !== previous?.submissionId)
|
||||
|| (current.submissionId !== null && current.submissionId !== previous?.submissionId
|
||||
&& current.submissionId !== previous?.steeringId)
|
||||
if (reading.pending && !ownInput) return
|
||||
content.current.applied = current
|
||||
if (current.ready && !content.current.opened) {
|
||||
|
||||
@@ -34,7 +34,6 @@ interface AssistantState {
|
||||
readonly firstVisibleSeq: number | undefined
|
||||
readonly firstVisibleTime: number | undefined
|
||||
readonly firstTokenTime: number | undefined
|
||||
readonly hidden: boolean
|
||||
readonly final: ConversationMatch | undefined
|
||||
readonly usage: unknown
|
||||
}
|
||||
@@ -48,7 +47,6 @@ function initialState(turn: number, step: number): AssistantState {
|
||||
firstVisibleSeq: undefined,
|
||||
firstVisibleTime: undefined,
|
||||
firstTokenTime: undefined,
|
||||
hidden: false,
|
||||
final: undefined,
|
||||
usage: undefined,
|
||||
}
|
||||
@@ -85,7 +83,6 @@ function resetForRetry(state: AssistantState): AssistantState {
|
||||
return {
|
||||
...initialState(state.turn, state.step),
|
||||
firstTokenTime: state.firstTokenTime,
|
||||
hidden: true,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -151,7 +148,6 @@ function updateChunk(
|
||||
...state,
|
||||
blocks,
|
||||
visibleBlocks,
|
||||
hidden: visibleBlocks > 0 ? false : state.hidden,
|
||||
...visibleBlocks > 0 && state.firstVisibleSeq === undefined
|
||||
? { firstVisibleSeq: seq, firstVisibleTime: time }
|
||||
: {},
|
||||
@@ -171,7 +167,6 @@ function settleMessage(
|
||||
...state,
|
||||
blocks,
|
||||
visibleBlocks: countVisibleBlocks(blocks),
|
||||
hidden: false,
|
||||
final: match,
|
||||
usage: event.data.usage,
|
||||
}
|
||||
@@ -289,7 +284,7 @@ function publishedAssistantData(
|
||||
return location?.kind === 'step' ? location.step.data.get('assistant-step') : undefined
|
||||
}
|
||||
|
||||
/** Per-step Assistant streaming/final/interruption Definition. */
|
||||
/** Per-step Assistant lifecycle; materialized keys survive cleared stream content as hidden Nodes. */
|
||||
export const assistantDefinition: ConversationNodeDefinition<AssistantState> = {
|
||||
kind: 'assistant-step',
|
||||
target: 'chat',
|
||||
@@ -337,16 +332,15 @@ export const assistantDefinition: ConversationNodeDefinition<AssistantState> = {
|
||||
}
|
||||
},
|
||||
buildViewNode: (context) => {
|
||||
const current = context.current.get('chat')
|
||||
const state = context.state ?? fallbackState(context)
|
||||
if (state === undefined) return null
|
||||
const data = publishedAssistantData(context)
|
||||
if (data === undefined) return null
|
||||
if (state === undefined || data === undefined) {
|
||||
return current == null ? null : { ...current, visibility: 'hidden' }
|
||||
}
|
||||
const settled = data.finalNode
|
||||
const visible = settled === undefined ? state.visibleBlocks > 0 : hasVisibleContent(data.blocks)
|
||||
if (settled === undefined && !visible) {
|
||||
const current = context.current.get('chat')
|
||||
if (!state.hidden || current === undefined || current === null) return null
|
||||
}
|
||||
if (settled === undefined && !visible && current == null) return null
|
||||
const anchorSeq = settled?.seq ?? state.firstVisibleSeq ?? context.matches[0]?.event.seq ?? 0
|
||||
return chatNode(context, 'assistant-step', anchorSeq, data, {
|
||||
visibility: settled?.interrupted === true || visible ? 'visible' : 'hidden',
|
||||
|
||||
@@ -1573,7 +1573,27 @@ describe('ChatView', () => {
|
||||
expect(view.getAllByText('即发即显')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('renders a local steer echo as pending steering before Host image admission completes', () => {
|
||||
it('does not pull the reader back to the tail when a local steer becomes Host-pending', () => {
|
||||
const h = makeHarness({ nodes: [assistant(1, 'working')] }, { running: true })
|
||||
const view = render(<h.ChatView {...h.props} />)
|
||||
const scroller = view.container.querySelector<HTMLElement>('[class*="scroll"]')!
|
||||
installScrollMetrics(scroller, 1_000, 300)
|
||||
act(() => { h.setSession({ pendingSubmissions: [{
|
||||
requestId: 'req-steer' as never, placement: 'steering', time: 5_000, text: 'continue here', attachments: [],
|
||||
}] }) })
|
||||
expect(scroller.scrollTop).toBe(700)
|
||||
readerScroll(scroller, 100)
|
||||
|
||||
act(() => { h.setSession({ testInbox: { 'next-turn': [], 'next-step': [{
|
||||
id: 'steer-occurrence' as never, role: 'user', source: { kind: 'user', rpcId: 'req-steer' as never },
|
||||
content: [{ type: 'text', text: 'continue here' }],
|
||||
}] } }) })
|
||||
expect(view.getAllByText('continue here')).toHaveLength(1)
|
||||
expect(scroller.scrollTop).toBe(100)
|
||||
expect(view.getByLabelText('回到底部')).toBeTruthy()
|
||||
})
|
||||
|
||||
it('retains the same local steer bubble through Host acceptance and claim until its durable node arrives', () => {
|
||||
const h = makeHarness(
|
||||
{ nodes: [assistant(1, 'working')] },
|
||||
{
|
||||
@@ -1604,8 +1624,20 @@ describe('ChatView', () => {
|
||||
})
|
||||
})
|
||||
expect(view.getAllByText('带图纠偏')).toHaveLength(1)
|
||||
expect(view.container.querySelector('[data-submission-echo]')).toBe(local)
|
||||
expect(view.container.querySelectorAll('[data-pending-steering]')).toHaveLength(1)
|
||||
|
||||
act(() => { h.setSession({ testInbox: { 'next-turn': [], 'next-step': [] } }) })
|
||||
expect(view.container.querySelector('[data-submission-echo]')).toBe(local)
|
||||
expect(view.getAllByText('带图纠偏')).toHaveLength(1)
|
||||
|
||||
act(() => { h.setChat({ nodes: [assistant(1, 'working'), {
|
||||
...steering(2, '带图纠偏', 1), source: { kind: 'user', rpcId: 'req-steer' as never },
|
||||
}] }) })
|
||||
expect(view.getAllByText('带图纠偏')).toHaveLength(1)
|
||||
expect(view.container.querySelector('[data-submission-echo]')).toBeNull()
|
||||
expect(view.container.querySelector('[data-pending-steering]')).not.toBeNull()
|
||||
act(() => { h.setSession({ pendingSubmissions: [] }) })
|
||||
expect(view.getAllByText('带图纠偏')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('keeps a queued echo out of the Chat flow before and after Host admission', () => {
|
||||
|
||||
@@ -993,6 +993,73 @@ describe('built-in conversation node Definitions', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it.each([
|
||||
{ outcome: 'attempt', startLoaded: true },
|
||||
{ outcome: 'attempt', startLoaded: false },
|
||||
{ outcome: 'abandoned', startLoaded: true },
|
||||
{ outcome: 'abandoned', startLoaded: false },
|
||||
])('retains the Assistant key after $outcome retirement (startLoaded=$startLoaded)', ({ outcome, startLoaded }) => {
|
||||
const attemptId = LlmAttemptId('retired-chat-attempt')
|
||||
const chunk = { type: 'text-delta' as const, index: 0, text: 'Discarded reply' }
|
||||
const stream = new AssistantStreamAccumulator()
|
||||
stream.push({ time: 1_030, chunk })
|
||||
const value = assembler([
|
||||
at(1, 'turn/start', { turn: 1 }),
|
||||
...startLoaded ? [at(2, 'step/start', { turn: 1, step: 1 })] : [],
|
||||
], !startLoaded)
|
||||
value.append({
|
||||
type: 'transient',
|
||||
event: {
|
||||
type: 'assistant/live-chunk', seq: 2.5, time: 1_030,
|
||||
data: { attemptId, turn: 1, step: 1, chunk },
|
||||
},
|
||||
})
|
||||
// The Step end can publish while the matching Assistant end frame still holds its settlement.
|
||||
if (!startLoaded) value.append(at(4, 'step/end', { turn: 1, step: 1 }))
|
||||
value.flush()
|
||||
const visible = node(snapshot(value), 'assistant-step')
|
||||
expect(visible?.visibility).toBe('visible')
|
||||
|
||||
if (outcome === 'attempt') {
|
||||
const event = at(3, 'assistant/attempt', { turn: 1, step: 1, stream: stream.snapshot() }).event
|
||||
if (event.type !== 'assistant/attempt') throw new Error('expected Assistant attempt')
|
||||
value.settleAssistant(attemptId, { type: 'event', event })
|
||||
} else value.settleAssistant(attemptId)
|
||||
expect(() => value.flush()).not.toThrow()
|
||||
const hidden = node(snapshot(value), 'assistant-step')
|
||||
expect(hidden?.key).toBe(visible?.key)
|
||||
expect(hidden?.visibility).toBe('hidden')
|
||||
expect(snapshot(value).order).not.toContain(hidden?.key)
|
||||
|
||||
if (!startLoaded) {
|
||||
value.append(at(5, 'turn/end', { turn: 1, reason: { kind: 'completed' } }))
|
||||
expect(() => value.flush()).not.toThrow()
|
||||
expect(node(snapshot(value), 'assistant-step')?.visibility).toBe('hidden')
|
||||
return
|
||||
}
|
||||
value.append(at(4, 'llm/retry', {
|
||||
retryId: 'retired-chat-retry', turn: 1, step: 1, provider: 'fake', mode: 'normal',
|
||||
policyKey: 'fake-normal', retry: 1, maxRetries: 2, delayMs: 10,
|
||||
failure: { code: 'TRANSPORT', message: 'temporary' },
|
||||
}))
|
||||
expect(() => value.flush()).not.toThrow()
|
||||
value.append({
|
||||
type: 'transient',
|
||||
event: {
|
||||
type: 'assistant/live-chunk', seq: 4.5, time: 1_050,
|
||||
data: {
|
||||
attemptId: LlmAttemptId('replacement-chat-attempt'), turn: 1, step: 1,
|
||||
chunk: { type: 'text-delta', index: 0, text: 'Replacement reply' },
|
||||
},
|
||||
},
|
||||
})
|
||||
value.flush()
|
||||
const replacement = node(snapshot(value), 'assistant-step')
|
||||
expect(replacement?.key).toBe(visible?.key)
|
||||
expect(replacement?.visibility).toBe('visible')
|
||||
expect(replacement?.data).toMatchObject({ blocks: [{ kind: 'text', text: 'Replacement reply' }] })
|
||||
})
|
||||
|
||||
it('omits first-token metrics after live settlement and after reopening the same history', () => {
|
||||
const attemptId = LlmAttemptId('settled-chat-timing')
|
||||
const starts = [
|
||||
|
||||
@@ -8,11 +8,11 @@
|
||||
- tab "Trajectory"
|
||||
- text: Reply with a one-sentence description of event sourcing, then stop. {{clock}}
|
||||
- button "Copy"
|
||||
- button "Thought for a while" [expanded]
|
||||
- button "Context injection runtime-context"
|
||||
- status: Worked
|
||||
- button "Took {{duration}}" [expanded]
|
||||
- group:
|
||||
- status: Retried model request (1/5) · {{duration}}
|
||||
- button "Think"
|
||||
- button "Analysis completed"
|
||||
- paragraph: Event sourcing is a pattern where all changes to an application's state are stored as an immutable, append-only sequence of events, rather than persisting only the current state, enabling full auditability, temporal queries, and event-driven architectures.
|
||||
- button "Copy"
|
||||
- button "Good response"
|
||||
|
||||
@@ -8,7 +8,8 @@
|
||||
- tab "Trajectory"
|
||||
- text: Reply with a one-sentence description of event sourcing, then stop. {{clock}}
|
||||
- button "Copy"
|
||||
- button "Thought for a while"
|
||||
- status: Worked
|
||||
- button "Took {{duration}}"
|
||||
- paragraph: Event sourcing is a pattern where all changes to an application's state are stored as an immutable, append-only sequence of events, rather than persisting only the current state, enabling full auditability, temporal queries, and event-driven architectures.
|
||||
- button "Copy"
|
||||
- button "Good response"
|
||||
|
||||
Reference in new issue
Block a user