diff --git a/apps/web/tests/steering.e2e.ts b/apps/web/tests/steering.e2e.ts index a687c45f57..451fa72221 100644 --- a/apps/web/tests/steering.e2e.ts +++ b/apps/web/tests/steering.e2e.ts @@ -7,10 +7,11 @@ import { fileURLToPath } from 'node:url' import { join } from 'node:path' import type { Browser, Page, WebSocketRoute } from 'playwright' import { chromium } from 'playwright' -import { afterAll, beforeAll, describe, expect, it, onTestFailed, type MockInstance } from 'vitest' +import { afterAll, beforeAll, describe, expect, it, onTestFailed, onTestFinished, type MockInstance } from 'vitest' +import type { Agent } from '@deepseek-ai/dsh-agent' import { parseSessionLog } from '@deepseek-ai/dsh-llm-replay' import type { SessionEvent } from '@deepseek-ai/dsh-session' -import { expandAssistantStream } from '@deepseek-ai/dsh-llm' +import { expandAssistantStream, type UserMessage } from '@deepseek-ai/dsh-llm' import { parseRemoteStreamServerMessage } from '@deepseek-ai/dsh-api-gateway/stream-protocol' import { acknowledgeReloadConnectionLoss, assertFixtureInventory, captureExpandedTurnProcessAria, captureStableAria, @@ -109,7 +110,7 @@ describe('web e2e: mid-turn steering lands durably and visibly', () => { } }) - it('strictly steers one queued row; the interjection is logged, rendered, and obeyed', async () => { + it('strictly steers one queued row and deduplicates its retry after claiming', async () => { onTestFailed(() => saveFailureShot(page, 'web-e2e-steering')) if (MODE !== 'record') { // The steer lands as a durable user/message, so the inventory holds @@ -155,12 +156,34 @@ describe('web e2e: mid-turn steering lands durably and visibly', () => { await compareOrRefreshGolden(MID_EXPECTED, snapshot, MODE) } - // Answer the composer; the tool result closes the step, the loop drains - // the steer as user/message, and the steered continuation runs the - // final model call. - await composer.getByRole('radio', { name: 'Yes' }).click() - await composer.getByRole('radio', { name: 'Yes' }).press('Enter') - await settled + const claimed = Promise.withResolvers<{ agent: Agent; message: UserMessage }>() + const release = Promise.withResolvers() + const stopHolding = scaffold.ctx.on('agent/pre-step', async ({ agent, messages }, next) => { + const message = messages.find(item => item.content.some(block => block.type === 'text' && block.text === STEER)) + if (message !== undefined) { + claimed.resolve({ agent, message }) + await release.promise + } + return next() + }) + onTestFinished(() => { release.resolve(undefined); stopHolding() }) + try { + await composer.getByRole('radio', { name: 'Yes' }).click() + await composer.getByRole('radio', { name: 'Yes' }).press('Enter') + const { agent, message } = await claimed.promise + expect(claimedMessages(sessionEvents, STEER)).toHaveLength(0) + const source = message.source + if (source.kind !== 'user' || !('rpcId' in source)) throw new Error('steering input has no request identity') + await expect(scaffold.ctx.sessionController.prompt({ + sessionId: agent.id, requestId: source.rpcId, mode: 'queue', + content: [{ type: 'text', text: STEER }], + }, new AbortController().signal)).resolves.toEqual({ accepted: true }) + expect([...agent.inbox.nextTurn, ...agent.inbox.nextStep]).toEqual([]) + } finally { + release.resolve(undefined) + stopHolding() + await settled + } if (MODE === 'record') { const sessionId = await settled diff --git a/packages/api/session-controller/README.i18n.yaml b/packages/api/session-controller/README.i18n.yaml index 2c2bbf17cc..3d31867b67 100644 --- a/packages/api/session-controller/README.i18n.yaml +++ b/packages/api/session-controller/README.i18n.yaml @@ -15,8 +15,8 @@ en: 1fdf707322fb6996 zh: 0204f7e9d4aa3cce /session-controller/use-this-package: - en: 425f679cb90f481b - zh: 15ed208a442b5b3b + en: 9cac36e3c0a32959 + zh: 37c42cc3edafa798 /session-controller/client-references: en: e8979a803ec3e194 zh: c567efcd971728ba diff --git a/packages/api/session-controller/README.md b/packages/api/session-controller/README.md index 3e58adb5ce..b430594b0e 100644 --- a/packages/api/session-controller/README.md +++ b/packages/api/session-controller/README.md @@ -27,6 +27,8 @@ Desktop analytics follows the [product collection policy](../../client/product-a ## Use this package +Prompt retries use the request identity in the first committed inbox insertion, including after the message is claimed or removed. Admission checks again after asynchronous attachment preparation, so concurrent retries accept one input. Acceptance acknowledges insertion, not execution or completion. + The controller reads a Session's recorded preset through the optional `agentPresets.presetForSession(session)` service method. A null result means no preset was recorded; an absent service means preset identity is unavailable to the controller. History pages and follow opening snapshots carry one `{ type: 'event', event: SessionWireEvent }` record per durable Session event. The Client retains each accepted record as one durable `SessionEventLikeEntry`; Assistant token boundaries remain inside the compact stream on `assistant/message` or `assistant/attempt`. Tool arguments, result content, failures, and `tool/result.data.meta` pass through unchanged; the controller does not resolve a Tool definition, run a presenter, or attach UI data. diff --git a/packages/api/session-controller/README.zh.md b/packages/api/session-controller/README.zh.md index 259c1be08d..314ec58840 100644 --- a/packages/api/session-controller/README.zh.md +++ b/packages/api/session-controller/README.zh.md @@ -27,6 +27,8 @@ kind: "package-reference" ## 使用本包 +提示重试按首次已提交入队记录中的请求标识去重，消息被领取或删除后仍然有效。异步附件准备结束后再次检查，因此并发重试只接受一条输入。接受确认入队，不代表执行或完成。 + 控制器通过可选服务方法 `agentPresets.presetForSession(session)` 读取 Session 已记录的预设。null 结果表示未记录预设；服务缺席表示控制器无法获知预设身份。 历史页与 follow opening 快照为每个持久 Session 事件携带一条 `{ type: 'event', event: SessionWireEvent }` record。Client 把每条已接受 record 保留为一个持久 `SessionEventLikeEntry`；Assistant token 边界保留在 `assistant/message` 或 `assistant/attempt` 的紧凑流内。工具参数、结果内容、失败信息和 `tool/result.data.meta` 原样通过；控制器不解析工具定义、不运行展示转换器，也不附加 UI 数据。 diff --git a/packages/api/session-controller/src/commands.ts b/packages/api/session-controller/src/commands.ts index 3e2d3e97f8..921d71cd23 100644 --- a/packages/api/session-controller/src/commands.ts +++ b/packages/api/session-controller/src/commands.ts @@ -17,10 +17,10 @@ import type {} from '@deepseek-ai/dsh-client-file-upload' import { ReasoningEffortId, createUserMessage, freezeMessage, } from '@deepseek-ai/dsh-llm' -import type { MessageSource } from '@deepseek-ai/dsh-llm' +import type { MessageId, MessageSource } from '@deepseek-ai/dsh-llm' import { buildForkSeed } from '@deepseek-ai/dsh-session/fork' import { SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session' -import type { SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session' +import type { Session, SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session' import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query' import { SessionTitleInvalidError } from '@deepseek-ai/dsh-session-title' import { assertNever } from '@deepseek-ai/dsh-util-values' @@ -86,6 +86,13 @@ function latestCompletedPrefixBoundary(events: readonly SessionEvent[]): Session /** Implements Session business commands delegated by the Session Controller Remote service. */ export class SessionCommandController { + /** First-insertion facts cached per exact Session; the log remains the recovery source. */ + private readonly promptReceipts = new WeakMap + requests: Set + }>() + private continuation!: ProjectionHandle /** Register complete-history empty-prompt eligibility. @returns when command admission is ready. */ @@ -313,7 +320,7 @@ export class SessionCommandController { */ async prompt(request: SessionPromptRequest, source: MessageSource): Promise { const agent = await this.resolveAgent(request.sessionId) - if (hasPromptRequest(agent, request.requestId)) return { accepted: true } + if (this.hasPromptRequest(agent.session, request.requestId)) return { accepted: true } if (request.content.length === 0) { try { return await agent.runMaintenance(() => { @@ -338,6 +345,7 @@ export class SessionCommandController { const hasImage = admission.content.some(part => part.type === 'image') const admit = async (): Promise => { try { + if (this.hasPromptRequest(agent.session, request.requestId)) return { accepted: true } if (hasImage) { const current = agent.model.current() if (current !== undefined) { @@ -352,6 +360,7 @@ export class SessionCommandController { } } const content = await this.ctx.attachments.admitPromptContent(admission.content) + if (this.hasPromptRequest(agent.session, request.requestId)) return { accepted: true } const message: UserMessage = createUserMessage({ content, source }) if (this.ctx.agents.get(agent.id) !== agent) { throw new RemoteError( @@ -365,6 +374,7 @@ export class SessionCommandController { else agent.followup(message) binding.commit() } catch (error) { + if (this.hasPromptRequest(agent.session, request.requestId)) return { accepted: true } if (remoteErrorOf(error) !== undefined) throw error if (error instanceof AttachmentError) { throw new RemoteError('session/attachment-invalid', error.message, { reason: error.code }) @@ -376,6 +386,32 @@ export class SessionCommandController { return hasImage ? this.agents.serializeImageAdmission(agent, admit) : admit() } + /** Recover first input acceptance, including claimed or removed messages, from unseen log entries. */ + private hasPromptRequest(session: Session, requestId: SessionRequestId): boolean { + let receipts = this.promptReceipts.get(session) + if (receipts === undefined) { + receipts = { offset: SessionLogOffset(0), messages: new Set(), requests: new Set() } + this.promptReceipts.set(session, receipts) + } + if (receipts.offset !== session.seq) { + // oxlint-disable-next-line typescript/no-deprecated -- Existing prompt retry history read, retained for acceptance recovery. + for (const event of session.snapshotEvents(receipts.offset)) { + if (event.type !== 'agent/inbox/spliced' && event.type !== 'user/message') continue + const messages = event.type === 'agent/inbox/spliced' ? event.data.inserted : [event.data] + for (const message of messages) { + if (receipts.messages.has(message.id)) continue + receipts.messages.add(message.id) + // A copied input is not a new request in the child, even if later reinserted there. + if (event.seq < session.inheritedEventCount) continue + const source = message.source + if ((source.kind === 'user' || source.kind === 'retry') && 'rpcId' in source) receipts.requests.add(source.rpcId) + } + } + receipts.offset = session.seq + } + return receipts.requests.has(requestId) + } + private async requireModel(selection: Pick): Promise { if (!await modelAvailable(this.ctx, selection)) { throw new RemoteError('session/model-unavailable', 'Select an available model before sending a message.', @@ -564,16 +600,3 @@ function resolvePromptFileReceipts( }) return { content: resolved, receiptIds: [...receiptIds] } } - -function hasPromptRequest(agent: Agent, requestId: SessionRequestId): boolean { - const matches = (message: UserMessage): boolean => { - const source = message.source - return (source.kind === 'user' || source.kind === 'retry') && 'rpcId' in source && source.rpcId === requestId - } - if (agent.inbox.nextTurn.some(matches) || agent.inbox.nextStep.some(matches)) return true - // oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred. - return agent.session.snapshotEvents().some((event) => { - if (event.type !== 'user/message') return false - return matches(event.data) - }) -} diff --git a/packages/api/session-controller/tests/commands-upload-file.host.spec.ts b/packages/api/session-controller/tests/commands-upload-file.host.spec.ts index 0f6aff10f2..89d4e35b5b 100644 --- a/packages/api/session-controller/tests/commands-upload-file.host.spec.ts +++ b/packages/api/session-controller/tests/commands-upload-file.host.spec.ts @@ -3,7 +3,7 @@ import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' import { Context, FiberState } from '@deepseek-ai/cordis' import AgentRegistry from '@deepseek-ai/dsh-agent' import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit' -import type { Agent } from '@deepseek-ai/dsh-agent' +import type { Agent, InboxTarget } from '@deepseek-ai/dsh-agent' import AttachmentStore, { AttachmentId } from '@deepseek-ai/dsh-attachment' import type { FileAttachmentRef, ImageAttachmentLimits, ImageAttachmentRef, SaveFileAttachment, SaveFileStreamAttachment, @@ -11,7 +11,7 @@ import type { } from '@deepseek-ai/dsh-attachment' import { createUserMessage } from '@deepseek-ai/dsh-llm' import type { MessageSource, UserMessage } from '@deepseek-ai/dsh-llm' -import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import SessionStore, { Session, SessionId, SessionLogOffset, type SessionEvent } from '@deepseek-ai/dsh-session' import CommandRuntime from '@deepseek-ai/dsh-commands' import { createScope } from '@deepseek-ai/dsh-scope' import FileUploads from '@deepseek-ai/dsh-client-file-upload' @@ -43,7 +43,11 @@ async function* chunks(...parts: readonly number[][]): AsyncIterable for (const part of parts) yield Uint8Array.from(part) } -async function uploadHarness(origin?: 'subagent', imageLimits: ImageAttachmentLimits = IMAGE_LIMITS): Promise<{ +async function uploadHarness( + origin?: 'subagent', + imageLimits: ImageAttachmentLimits = IMAGE_LIMITS, + seed?: readonly SessionEvent[], +): Promise<{ ctx: Context controller: SessionCommandController uploads: FileUploads @@ -63,10 +67,16 @@ async function uploadHarness(origin?: 'subagent', imageLimits: ImageAttachmentLi await ctx.plugin(AgentRegistry) await ctx.plugin(CommandRuntime) const session = ctx.sessions.create(SESSION, { - meta: { cwd: '/workspace', ...(origin === undefined ? {} : { origin }) }, + ...(seed === undefined ? {} : { seed, inheritedEventCount: SessionLogOffset(seed.length) }), + meta: { cwd: '/workspace', ...(origin === undefined ? {} : { origin }), ...(seed === undefined ? {} : { isSeeded: true }) }, }) const inbox = createInboxStub() - const followup = vi.fn() + const accept = (target: InboxTarget, message: UserMessage): void => { + const pending = target === 'next-turn' ? inbox.nextTurn : inbox.nextStep + session.append('agent/inbox/spliced', { target, start: pending.length, inserted: [message] }) + inbox.append(target, message) + } + const followup = vi.fn((message: UserMessage) => { accept('next-turn', message) }) const options = { provider: 'fixture', model: 'fixture-model' } const projection = await ctx.sessionProjections.register(modelSelectionProjectionDefinition) const agent = { @@ -81,7 +91,7 @@ async function uploadHarness(origin?: 'subagent', imageLimits: ImageAttachmentLi inbox, status: 'idle', ctx, - steer: vi.fn(), + steer: vi.fn((message: UserMessage) => { accept('next-step', message) }), followup, cancel: vi.fn(), } satisfies Agent @@ -524,14 +534,117 @@ describe('Session file uploads', () => { }) it('deduplicates a retried rpcId already present in the Agent inbox', async () => { - const { controller, agent, followup } = await uploadHarness() + const { controller, followup } = await uploadHarness() const request = promptRequest([{ type: 'text', text: 'once' }]) await controller.prompt(request, USER_SOURCE) - agent.inbox.append('next-turn', followup.mock.calls[0]?.[0] as UserMessage) await expect(controller.prompt(request, USER_SOURCE)).resolves.toEqual({ accepted: true }) expect(followup).toHaveBeenCalledOnce() }) + it.each(['claimed', 'removed'] as const)('deduplicates a %s input before user/message is logged', async (state) => { + const { controller, uploads, agent, followup } = await uploadHarness() + const file = await uploads.upload(agent, { data: 'AAAA' }, new AbortController().signal) + const request = promptRequest([{ type: 'file', receiptId: file.receiptId }]) + await controller.prompt(request, USER_SOURCE) + if (state === 'removed') { + await controller.updateQueue({ sessionId: agent.id, itemId: agent.inbox.nextTurn[0]!.id, action: { kind: 'remove' } }) + expect(uploads.resolve(agent, file.receiptId)).toBeUndefined() + } else { + agent.session.append('agent/inbox/spliced', { target: 'next-turn', start: 0, removedCount: 1, inserted: [] }) + agent.inbox.clear() + } + expect(agent.inbox.nextTurn).toEqual([]) + await expect(controller.prompt(request, USER_SOURCE)).resolves.toEqual({ accepted: true }) + expect(followup).toHaveBeenCalledOnce() + }) + + it('does not reuse inherited acceptance when a copied message is reinserted in the child', async () => { + const ancestor = Session.create(SessionId('ancestor')) + const message = createUserMessage({ content: [{ type: 'text', text: 'parent input' }], source: USER_SOURCE }) + ancestor.append('agent/inbox/spliced', { target: 'next-turn', start: 0, inserted: [message] }) + ancestor.append('agent/inbox/spliced', { target: 'next-turn', start: 0, removedCount: 1, inserted: [] }) + const { controller, agent, followup } = await uploadHarness(undefined, IMAGE_LIMITS, ancestor.snapshotEvents()) + agent.session.append('agent/inbox/spliced', { target: 'next-step', start: 0, inserted: [message] }) + const request = promptRequest([{ type: 'text', text: 'child input' }]) + await controller.prompt(request, USER_SOURCE) + await controller.prompt(request, USER_SOURCE) + expect(followup).toHaveBeenCalledOnce() + expect(agent.inbox.nextTurn[0]?.id).not.toBe(message.id) + }) + + it('recovers acceptance from an insertion already removed before the first retry lookup', async () => { + const { controller, agent, followup } = await uploadHarness() + const message = createUserMessage({ content: [{ type: 'text', text: 'accepted' }], source: USER_SOURCE }) + agent.session.append('agent/inbox/spliced', { target: 'next-turn', start: 0, inserted: [message] }) + agent.session.append('agent/inbox/spliced', { target: 'next-turn', start: 0, removedCount: 1, inserted: [], outcome: 'canceled' }) + await expect(controller.prompt(promptRequest([{ type: 'text', text: 'retry' }]), USER_SOURCE)).resolves.toEqual({ accepted: true }) + expect(followup).not.toHaveBeenCalled() + }) + + it.each(['queue', 'steer'] as const)('accepts concurrent %s retries only once after asynchronous preparation', async (mode) => { + const { ctx, controller, agent, followup } = await uploadHarness() + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + let calls = 0 + const admit = ctx.attachments.admitPromptContent.bind(ctx.attachments) + vi.spyOn(ctx.attachments, 'admitPromptContent').mockImplementation(async (content) => { + if (++calls === 2) entered.resolve(undefined) + await release.promise + return admit(content) + }) + const request = { ...promptRequest([{ type: 'text', text: 'once' }]), mode } + const prompting = Promise.all([controller.prompt(request, USER_SOURCE), controller.prompt(request, USER_SOURCE)]) + onTestFinished(async () => { + release.resolve(undefined) + await Promise.allSettled([prompting]) + }) + await entered.promise + release.resolve(undefined) + await expect(prompting).resolves.toEqual([{ accepted: true }, { accepted: true }]) + expect(followup).toHaveBeenCalledTimes(mode === 'queue' ? 1 : 0) + expect(agent.inbox.nextTurn).toHaveLength(mode === 'queue' ? 1 : 0) + expect(agent.inbox.nextStep).toHaveLength(mode === 'steer' ? 1 : 0) + }) + + it('skips image preparation for a serialized retry already accepted', async () => { + const { controller, followup, saveImages } = await uploadHarness() + const release = Promise.withResolvers() + saveImages.mockReturnValueOnce(release.promise) + const request = promptRequest([{ type: 'image', mediaType: 'image/png', data: 'AAAA' }]) + const prompting = Promise.all([controller.prompt(request, USER_SOURCE), controller.prompt(request, USER_SOURCE)]) + onTestFinished(async () => { + release.resolve([STAGED_IMAGE]) + await Promise.allSettled([prompting]) + }) + await vi.waitFor(() => { expect(saveImages).toHaveBeenCalledOnce() }) + release.resolve([STAGED_IMAGE]) + await expect(prompting).resolves.toEqual([{ accepted: true }, { accepted: true }]) + expect(saveImages).toHaveBeenCalledOnce() + expect(followup).toHaveBeenCalledOnce() + }) + + it('acknowledges acceptance when concurrent preparation fails after another call accepts', async () => { + const { ctx, controller, followup } = await uploadHarness() + const entered = Promise.withResolvers() + const release = Promise.withResolvers() + vi.spyOn(ctx.attachments, 'admitPromptContent').mockImplementationOnce(async () => { + entered.resolve(undefined) + await release.promise + throw new Error('preparation failed') + }) + const request = promptRequest([{ type: 'text', text: 'once' }]) + const prompting = controller.prompt(request, USER_SOURCE) + onTestFinished(async () => { + release.resolve(undefined) + await Promise.allSettled([prompting]) + }) + await entered.promise + await expect(controller.prompt(request, USER_SOURCE)).resolves.toEqual({ accepted: true }) + release.resolve(undefined) + await expect(prompting).resolves.toEqual({ accepted: true }) + expect(followup).toHaveBeenCalledOnce() + }) + it('deduplicates a retried rpcId already present in the durable log', async () => { const { controller, agent, followup } = await uploadHarness() const request = promptRequest([{ type: 'text', text: 'once' }]) diff --git a/packages/api/session-controller/tests/empty-prompt.host.spec.ts b/packages/api/session-controller/tests/empty-prompt.host.spec.ts index 4f38b9901b..778b3fdee5 100644 --- a/packages/api/session-controller/tests/empty-prompt.host.spec.ts +++ b/packages/api/session-controller/tests/empty-prompt.host.spec.ts @@ -94,3 +94,30 @@ it('consumes injected context with the empty prompt', async () => { expect(agent.inbox.nextStep).toHaveLength(0) expect(ctx.sessionProjections.snapshot(agent.session).values.conversationContinuation).not.toBeNull() }) + +it('deduplicates empty continuation after claim and before user-message admission', async () => { + const { ctx, remote, agent, adapter, request } = await setup() + agent.followup(createUserMessage({ content: [{ type: 'text', text: 'original task' }], source: { kind: 'user' } })) + await agent.whenIdle() + const claimed = Promise.withResolvers() + const release = Promise.withResolvers() + onTestFinished(() => { release.resolve(undefined) }) + ctx.on('agent/pre-step', async ({ messages }, next) => { + if (messages.some(message => message.source.kind === 'retry')) { + claimed.resolve(undefined) + await release.promise + } + return next() + }) + const continuation = request('continue-once') + expect((await remote.prompt(continuation)).ok).toBe(true) + await claimed.promise + expect(agent.inbox.nextTurn).toEqual([]) + expect(agent.session.snapshotEvents().some(event => event.type === 'user/message' && event.data.source.kind === 'retry')).toBe(false) + expect((await remote.prompt(continuation)).ok).toBe(true) + release.resolve(undefined) + await agent.whenIdle() + expect(adapter.requests).toHaveLength(2) + expect(agent.session.snapshotEvents().flatMap(event => event.type === 'agent/inbox/spliced' ? event.data.inserted : []) + .filter(message => message.source.kind === 'retry')).toHaveLength(1) +})