diff --git a/packages/api/session-controller/tests/client-contract.client.spec.ts b/packages/api/session-controller/tests/client-contract.client.spec.ts index b2abdc2dc2..44f2373ea6 100644 --- a/packages/api/session-controller/tests/client-contract.client.spec.ts +++ b/packages/api/session-controller/tests/client-contract.client.spec.ts @@ -165,6 +165,17 @@ describe('Client Session contracts', () => { kind: 'settle-assistant', attemptId: 'attempt-two', }) + + // A settlement past every current entry appends at the tail instead of splicing before one. + const tail = assistantSettlement(SessionSeq(4)) + feed.append(transient('attempt-three', 3.75)) + feed.settleAssistant(LlmAttemptId('attempt-three'), tail) + expect(feed.getSnapshot().entries).toEqual([opening, settlement, later, tail]) + expect(feed.getSnapshot().change).toEqual({ + kind: 'settle-assistant', + attemptId: 'attempt-three', + entry: tail, + }) }) }) diff --git a/packages/api/session-controller/tests/commands-create-fork.host.spec.ts b/packages/api/session-controller/tests/commands-create-fork.host.spec.ts index eb2093ea18..818abbc021 100644 --- a/packages/api/session-controller/tests/commands-create-fork.host.spec.ts +++ b/packages/api/session-controller/tests/commands-create-fork.host.spec.ts @@ -4,7 +4,8 @@ import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' import type { Agent, AgentHandle, CreateAgentOptions } from '@deepseek-ai/dsh-agent' import type {} from '@deepseek-ai/dsh-agent-preset-registry' import { createUserMessage } from '@deepseek-ai/dsh-llm' -import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import SessionStore, { SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session' +import { SessionQueryError } from '@deepseek-ai/dsh-session-query' import { RemoteError } from '@deepseek-ai/dsh-typert-protocol' import TypertRegistry from '@deepseek-ai/dsh-typert-registry' import type { Workspace, WorkspaceId } from '@deepseek-ai/dsh-workspace' @@ -176,6 +177,23 @@ function completedSession( return session } +/** + * A fork creation outcome that stores the seeded child the way the Agent factory would, so a later + * request can observe it under the id the caller named. + * @param ctx - context owning the Session store. + * @returns the mocked `ctx.agents.create` implementation. + */ +function seededChildCreate(ctx: Context) { + return vi.fn((options: CreateAgentOptions): Promise => { + ctx.sessions.create(options.sessionId, { + ...(options.seed === undefined ? {} : { seed: options.seed }), + ...(options.inheritedEventCount === undefined ? {} : { inheritedEventCount: options.inheritedEventCount }), + ...(options.meta === undefined ? {} : { meta: options.meta }), + }) + return Promise.resolve(resolvedHandle(ctx, options.sessionId)) + }) +} + function resolvedHandle(ctx: Context, sessionId: SessionId): AgentHandle { return { agent: { id: sessionId, status: 'idle', ctx } as Agent, @@ -295,3 +313,86 @@ describe('Session fork failures', () => { await ctx.fiber.dispose() }) }) +describe('Session fork named children', () => { + it('adopts the child an earlier named fork already materialized', async () => { + const ctx = await baseContext() + ctx.provide('workspaceRegistry', { list: () => [] } as never) + const source = completedSession(ctx, 'named-source') + const create = seededChildCreate(ctx) + vi.spyOn(ctx.agents, 'create').mockImplementation(create) + const controller = new SessionCommandController(ctx, controllerAgents(ctx), '/default') + const child = SessionId('named-child') + + // Nothing is stored under the name yet, so the first call forks the child it was asked for. + await expect(controller.fork({ sessionId: source.id, childSessionId: child })) + .resolves.toEqual({ sessionId: child }) + expect(create).toHaveBeenCalledOnce() + + // The retry reads the stored child as this request's own fork and answers with it, without + // materializing a second child. + await expect(controller.fork({ sessionId: source.id, childSessionId: child })) + .resolves.toEqual({ sessionId: child }) + expect(create).toHaveBeenCalledOnce() + await ctx.fiber.dispose() + }) + + it('refuses a named child another Session holds, and one seeded at another cut', async () => { + const foreign = await baseContext() + foreign.provide('workspaceRegistry', { list: () => [] } as never) + const source = completedSession(foreign, 'named-foreign-source') + // The name is taken by a Session that was not seeded from this source at all. + foreign.sessions.create(SessionId('foreign-child'), { + meta: { parentSession: SessionId('another-source') }, + }) + const foreignController = new SessionCommandController(foreign, controllerAgents(foreign), '/default') + await expect(foreignController.fork({ sessionId: source.id, childSessionId: SessionId('foreign-child') })) + .rejects.toMatchObject({ code: 'session/fork-conflict' }) + await foreign.fiber.dispose() + + const cut = await baseContext() + cut.provide('workspaceRegistry', { list: () => [] } as never) + const cutSource = completedSession(cut, 'named-cut-source') + // Same source, but the stored child inherited a shorter prefix than this fork would write. + cut.sessions.create(SessionId('wrong-cut-child'), { + seed: cutSource.snapshotEvents().slice(0, 2), + inheritedEventCount: SessionLogOffset(2), + meta: { parentSession: cutSource.id, isSeeded: true }, + }) + const cutController = new SessionCommandController(cut, controllerAgents(cut), '/default') + await expect(cutController.fork({ sessionId: cutSource.id, childSessionId: SessionId('wrong-cut-child') })) + .rejects.toMatchObject({ code: 'session/fork-conflict' }) + await cut.fiber.dispose() + }) + + it('maps a named child it cannot read to an internal fork error', async () => { + const corrupt = await baseContext() + corrupt.provide('workspaceRegistry', { list: () => [] } as never) + const corruptSource = completedSession(corrupt, 'named-corrupt-source') + const corruptQuery = corrupt.sessionQuery + const observeCorrupt = corruptQuery.observeSession.bind(corruptQuery) + vi.spyOn(corruptQuery, 'observeSession').mockImplementation(async (sessionId, options) => ( + sessionId === corruptSource.id + ? observeCorrupt(sessionId, options) + : Promise.reject(new SessionQueryError('child is corrupt', 'SESSION_QUERY_CORRUPT_SESSION')) + )) + const corruptController = new SessionCommandController(corrupt, controllerAgents(corrupt), '/default') + await expect(corruptController.fork({ sessionId: corruptSource.id, childSessionId: SessionId('corrupt-child') })) + .rejects.toMatchObject({ code: 'gateway/internal' }) + await corrupt.fiber.dispose() + + const offline = await baseContext() + offline.provide('workspaceRegistry', { list: () => [] } as never) + const offlineSource = completedSession(offline, 'named-offline-source') + const offlineQuery = offline.sessionQuery + const observeOffline = offlineQuery.observeSession.bind(offlineQuery) + vi.spyOn(offlineQuery, 'observeSession').mockImplementation(async (sessionId, options) => ( + sessionId === offlineSource.id + ? observeOffline(sessionId, options) + : Promise.reject(new Error('storage offline')) + )) + const offlineController = new SessionCommandController(offline, controllerAgents(offline), '/default') + await expect(offlineController.fork({ sessionId: offlineSource.id, childSessionId: SessionId('offline-child') })) + .rejects.toMatchObject({ code: 'gateway/internal' }) + await offline.fiber.dispose() + }) +}) diff --git a/packages/api/session-controller/tests/remote-access.host.spec.ts b/packages/api/session-controller/tests/remote-access.host.spec.ts index e45dfbd3b4..3bb54de7bb 100644 --- a/packages/api/session-controller/tests/remote-access.host.spec.ts +++ b/packages/api/session-controller/tests/remote-access.host.spec.ts @@ -9,7 +9,9 @@ import { Context } from '@deepseek-ai/cordis' import type { Agent } from '@deepseek-ai/dsh-agent' -import { mountAgentLoopTestDependencies, mountAgentLoopTestHarness } from '@deepseek-ai/dsh-agent-loop-testkit' +import { + mountAgentLoopTestDependencies, mountAgentLoopTestHarness, type AgentLoopTestHarness, +} from '@deepseek-ai/dsh-agent-loop-testkit' import TypertGatewayService from '@deepseek-ai/dsh-api-gateway' import type { RemoteAccessBinding, RemoteAccessRole } from '@deepseek-ai/dsh-api-remote-access' import { createUserMessage } from '@deepseek-ai/dsh-llm' @@ -19,7 +21,9 @@ import TypertRegistry from '@deepseek-ai/dsh-typert-registry' import { afterEach, describe, expect, it, vi } from 'vitest' import type SessionController from '../src/index.ts' import { ApiSessionList } from '../src/list.ts' -import type { SessionControlFrame, SessionRequestId, SessionSearchValue } from '../src/types.ts' +import type { + SessionControlFrame, SessionPromptRequest, SessionRequestId, SessionSearchValue, +} from '../src/types.ts' import { createSessionTestController } from './test-remote.ts' declare module '@deepseek-ai/dsh-typert-protocol' { @@ -83,6 +87,8 @@ async function harness(): Promise<{ readonly controller: SessionController readonly shared: Agent readonly hidden: Agent + /** Admit pending input through the production loop driver, as a running turn would. */ + readonly claim: AgentLoopTestHarness['claim'] /** The Context a Gateway dispatch of `session/` binds for `principal`. */ readonly as: (principal: Principal, method: string) => Context }> { @@ -104,13 +110,40 @@ async function harness(): Promise<{ } return ctx.extend({ invocation }) } - return { ctx, controller, shared, hidden, as } + return { ctx, controller, shared, hidden, claim: (...args: Parameters) => loop.claim(...args), as } } function message(text: string) { return createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }) } +/** + * One queued prompt from a caller: the request identity is the whole difference between retries. + * @param requestId - repeated request identity. + * @returns the prompt request. + */ +function promptRequest(requestId: string): SessionPromptRequest { + return { + sessionId: SHARED, + requestId: requestId as SessionRequestId, + mode: 'queue', + content: [{ type: 'text', text: requestId }], + } +} + +/** + * Durable messages one request identity delivered into a Session's inbox splices. + * @param agent - Session whose durable log is read. + * @param requestId - repeated request identity. + * @returns the matching user messages. + */ +function deliveredFor(agent: Agent, requestId: string) { + return agent.session.snapshotEvents() + .flatMap(event => event.type === 'agent/inbox/spliced' ? event.data.inserted : []) + .filter(message => message.source.kind === 'user' + && 'rpcId' in message.source && message.source.rpcId === requestId) +} + describe('Session Controller access', () => { it('owns the Session resource kind and its default sessionId field', async () => { const { ctx } = await harness() @@ -215,6 +248,112 @@ describe('Session Controller access', () => { ]) }) + it('refuses a prompt whose Session grant is withdrawn during admission', async () => { + const { shared, as } = await harness() + // The Gateway admitted the call earlier; the controller re-reads the caller's grant after every + // await and immediately before delivery, so a share narrowed meanwhile must not deliver. + const signal = new AbortController().signal + await expect(as(GUEST, 'prompt').sessionController.prompt(promptRequest('withdrawn-1'), signal)) + .rejects.toMatchObject({ + code: 'remote-access/denied', + details: { action: 'session/prompt', reason: 'access-withdrawn' }, + }) + expect(shared.inbox.nextTurn).toEqual([]) + }) + + it('cancels admission when the caller aborts after the prompt was admitted', async () => { + const { ctx, shared, as } = await harness() + guestPolicy(ctx, DRIVES_SHARED) + const attachments = ctx.attachments + const admit = attachments.admitPromptContent.bind(attachments) + let release!: () => void + const held = new Promise((resolve) => { release = resolve }) + const admitted = vi.spyOn(attachments, 'admitPromptContent').mockImplementationOnce(async (content) => { + await held + return admit(content) + }) + const abort = new AbortController() + const pending = as(GUEST, 'prompt').sessionController.prompt(promptRequest('aborted-1'), abort.signal) + await vi.waitFor(() => { expect(admitted).toHaveBeenCalledOnce() }) + abort.abort() + release() + await expect(pending).rejects.toMatchObject({ code: 'gateway/cancelled' }) + expect(shared.inbox.nextTurn).toEqual([]) + }) + + it('shares one in-flight admission and refuses a duplicate requestId from another caller', async () => { + const { ctx, shared, as } = await harness() + guestPolicy(ctx, DRIVES_SHARED) + const attachments = ctx.attachments + const admit = attachments.admitPromptContent.bind(attachments) + const gates: (() => void)[] = [] + vi.spyOn(attachments, 'admitPromptContent').mockImplementation(async (content) => { + await new Promise((resolve) => { gates.push(resolve) }) + return admit(content) + }) + const signal = new AbortController().signal + + // A duplicate from the same caller arrives while the first admission is in flight: it shares that + // admission instead of starting a second one. + const first = as(GUEST, 'prompt').sessionController.prompt(promptRequest('inflight-shared'), signal) + await vi.waitFor(() => { expect(gates).toHaveLength(1) }) + const duplicate = as(GUEST, 'prompt').sessionController.prompt(promptRequest('inflight-shared'), signal) + await vi.waitFor(() => { expect(gates).toHaveLength(1) }) + gates[0]?.() + await expect(first).resolves.toEqual({ accepted: true }) + await expect(duplicate).resolves.toEqual({ accepted: true }) + expect(deliveredFor(shared, 'inflight-shared')).toHaveLength(1) + + // A duplicate from another caller while that admission is still pending is a conflict. + const other = as(OWNER, 'prompt').sessionController.prompt(promptRequest('inflight-conflict'), signal) + await vi.waitFor(() => { expect(gates).toHaveLength(2) }) + await expect(as(GUEST, 'prompt').sessionController.prompt(promptRequest('inflight-conflict'), signal)) + .rejects.toMatchObject({ + code: 'session/request-conflict', + details: { requestId: 'inflight-conflict' }, + }) + gates[1]?.() + await expect(other).resolves.toEqual({ accepted: true }) + }) + + it('reads a retried requestId from the drained splice and refuses it from another caller', async () => { + const { ctx, shared, as, claim } = await harness() + guestPolicy(ctx, DRIVES_SHARED) + const signal = new AbortController().signal + + await expect(as(GUEST, 'prompt').sessionController.prompt(promptRequest('retried-1'), signal)) + .resolves.toEqual({ accepted: true }) + // Drain the pending input as a running turn would: the requestId now lives only in the durable splice. + claim(shared, 'next-turn', 1) + expect(shared.inbox.nextTurn).toEqual([]) + // The same caller's retry reads the accepted message back from that splice. + await expect(as(GUEST, 'prompt').sessionController.prompt(promptRequest('retried-1'), signal)) + .resolves.toEqual({ accepted: true }) + // Another caller repeating the identity is refused as a conflict, with no second delivery. + await expect(as(OWNER, 'prompt').sessionController.prompt(promptRequest('retried-1'), signal)) + .rejects.toMatchObject({ + code: 'session/request-conflict', + details: { requestId: 'retried-1' }, + }) + expect(deliveredFor(shared, 'retried-1')).toHaveLength(1) + }) + + it('refuses cancellation once the caller no longer holds the Session', async () => { + const { ctx, as } = await harness() + // The call still reaches the controller; the Session role the page holds no longer admits cancel. + guestPolicy(ctx, READS_SHARED) + let failure: unknown + try { + as(GUEST, 'cancel').sessionController.cancel({ sessionId: SHARED }) + } catch (error) { + failure = error + } + expect(failure).toMatchObject({ + code: 'remote-access/denied', + details: { action: 'session/cancel', reason: 'access-withdrawn' }, + }) + }) + it('lets a non-owner read projections only in a Session its Session role names', async () => { const { ctx } = await harness() await ctx.plugin(TypertGatewayService)