refactor(api-session): remove interaction transport

This commit is contained in:
imccyu committed 2026-08-23 16:16:03 +08:00
1 parent e8ede58603
commit 639dcef5d8
21 files changed
+8 -1269

No files matched your search

+1 -6
View File
@@ -98,8 +98,6 @@
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-typert-protocol": "workspace:^",
"@deepseek-ai/dsh-typert-registry": "workspace:^",
"@deepseek-ai/dsh-user-approval": "workspace:^",
"@deepseek-ai/dsh-user-questions": "workspace:^",
"@deepseek-ai/dsh-workspace": "workspace:^"
},
"peerDependenciesMeta": {
@@ -107,8 +105,7 @@
"@deepseek-ai/dsh-session-persistence": { "optional": true },
"@deepseek-ai/dsh-session-projection": { "optional": true },
"@deepseek-ai/dsh-session-projection-cache": { "optional": true },
"@deepseek-ai/dsh-tools": { "optional": true },
"@deepseek-ai/dsh-user-approval": { "optional": true }
"@deepseek-ai/dsh-tools": { "optional": true }
},
"devDependencies": {
"@deepseek-ai/cordis": "workspace:^",
@@ -133,8 +130,6 @@
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-typert-protocol": "workspace:^",
"@deepseek-ai/dsh-typert-registry": "workspace:^",
"@deepseek-ai/dsh-user-approval": "workspace:^",
"@deepseek-ai/dsh-user-questions": "workspace:^",
"@deepseek-ai/dsh-workspace": "workspace:^"
}
}
+3 -310
View File
@@ -1,6 +1,5 @@
/** Live Session control state, interaction waits, and reconnect baselines. */
/** Live Session queue, jobs, and projection state with reconnect baselines. */
import { randomUUID } from 'node:crypto'
import type { Context } from '@deepseek-ai/cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import type { JobSnapshot } from '@deepseek-ai/dsh-jobs'
@@ -8,59 +7,25 @@ import type {
JsonValue, Session, SessionEvent, SessionEventMap, SessionId, UserMessage,
} from '@deepseek-ai/dsh-session'
import type {
ApprovalOutcome, ApprovalRequestEvent, ApprovalRequestId,
} from '@deepseek-ai/dsh-user-approval/types'
import {
UserQuestionError,
type AskUserQuestionAnswer,
type AskUserQuestionItem,
type AskUserQuestionRequest,
} from '@deepseek-ai/dsh-user-questions'
import type {
SessionApprovalRequest,
SessionApprovalResponse,
SessionControlBaseline,
SessionControlFrame,
SessionInteractionId,
SessionJob,
SessionProjectionsBlock,
SessionProjectionValues,
SessionQuestionRequest,
SessionQuestionResponse,
SessionQueuedItem,
SessionRespondReceipt,
SessionRespondRequest,
} from './types.ts'
interface PendingApproval extends SessionApprovalRequest {
readonly signal?: AbortSignal
readonly onAbort?: () => void
settle(outcome: ApprovalOutcome): void
}
interface PendingQuestion extends SessionQuestionRequest {
readonly questions: AskUserQuestionItem[]
readonly signal?: AbortSignal
onAbort?: () => void
resolve(answer: AskUserQuestionAnswer): void
reject(error: UserQuestionError): void
}
/** Owns the Host-wide control stream and answerable interaction registry. */
/** Owns the Host-wide Session control stream. */
export class SessionControlController {
private readonly streams = new Set<ControlQueue>()
private readonly approvals = new Map<SessionInteractionId, PendingApproval>()
private readonly questions = new Map<SessionInteractionId, PendingQuestion>()
/** @param ctx - Host context carrying live Agent, projection, jobs, approval, and question services. */
/** @param ctx - Host context carrying live Agent, projection, and jobs services. */
constructor(private readonly ctx: Context) {
ctx.on('session/event', (session, event) => { this.onSessionEvent(session, event) })
ctx.on('session/created', (session) => {
const jobs = this.jobsFor(this.ctx.agents.get(session.id))
if (jobs.length > 0) this.broadcast({ type: 'jobs', sessionId: session.id, jobs })
})
ctx.on('session/disposed', (session) => { this.cancelSessionInteractions(session.id) })
ctx.inject(['sessionProjections'], (projectionCtx) => {
projectionCtx.sessionProjections.onChanged((session, key, value, seq) => {
this.broadcast({
@@ -75,23 +40,7 @@ export class SessionControlController {
ctx.inject(['jobs'], (jobsCtx) => {
jobsCtx.jobs.onJobsChanged((owner) => { this.onJobsChanged(owner) })
})
ctx.inject(['approval'], (approvalCtx) => {
approvalCtx.on('approval/request', (request, next) => this.requestApproval(request, next))
})
const disposeQuestions = ctx.userQuestions.registerProvider({
ask: request => this.requestQuestion(request),
})
ctx.effect(() => () => {
disposeQuestions()
for (const pending of [...this.questions.values()]) {
this.claimQuestion(pending, 'cancelled')
pending.reject(new UserQuestionError(
'Session Controller user-questions provider was disposed',
'ASK_ABORTED',
))
}
for (const pending of [...this.approvals.values()]) pending.settle('cancelled')
for (const stream of this.streams) stream.end()
this.streams.clear()
}, 'session-controller.control')
@@ -115,19 +64,6 @@ export class SessionControlController {
}
}
/**
* Settle one still-pending approval or question.
* @param request - interaction identity and caller response.
* @returns whether a matching pending interaction accepted the response.
*/
respond(request: SessionRespondRequest): SessionRespondReceipt {
const approval = this.approvals.get(request.interactionId)
if (approval !== undefined) return this.respondApproval(approval, request)
const question = this.questions.get(request.interactionId)
if (question !== undefined) return this.respondQuestion(question, request)
return { accepted: false, reason: 'not-pending' }
}
private baseline(): SessionControlBaseline {
const sessions = this.ctx.sessions.list()
const queues = Object.create(null) as Record<SessionId, readonly SessionQueuedItem[]>
@@ -140,8 +76,6 @@ export class SessionControlController {
return {
queues,
jobs,
approvals: [...this.approvals.values()].map(pending => approvalRequest(pending)),
questions: [...this.questions.values()].map(pending => questionRequest(pending)),
projections: this.projectionBaseline(sessions),
}
}
@@ -194,161 +128,6 @@ export class SessionControlController {
return jobs === undefined ? [] : jobs.list(agent).map(jobView)
}
private requestQuestion(request: AskUserQuestionRequest): Promise<AskUserQuestionAnswer> {
const sessionId = request.agent?.id
if (sessionId === undefined) {
return Promise.reject(new UserQuestionError(
'web user interaction requires an agent-owned session',
'ASK_MISSING_AGENT',
))
}
if (request.signal?.aborted === true) {
return Promise.reject(new UserQuestionError(
'ask_user_question was aborted before the user answered',
'ASK_ABORTED',
))
}
return new Promise<AskUserQuestionAnswer>((resolve, reject) => {
const interactionId = newInteractionId()
const pending: PendingQuestion = {
interactionId,
sessionId,
questions: request.questions,
resolve,
reject,
...(request.signal === undefined ? {} : { signal: request.signal }),
}
const onAbort = (): void => {
if (!this.claimQuestion(pending, 'cancelled')) return
reject(new UserQuestionError(
'ask_user_question was aborted before the user answered',
'ASK_ABORTED',
))
}
pending.onAbort = onAbort
this.questions.set(interactionId, pending)
request.signal?.addEventListener('abort', onAbort, { once: true })
if (request.signal?.aborted === true) {
onAbort()
return
}
this.broadcast({ type: 'question/requested', ...questionRequest(pending) })
})
}
private requestApproval(
request: ApprovalRequestEvent,
next: () => Promise<ApprovalOutcome>,
): Promise<ApprovalOutcome> {
if (request.signal?.aborted === true) return Promise.resolve('cancelled')
const approvalId = findApprovalId(request, this.approvals.values())
if (approvalId === undefined) return next()
return new Promise<ApprovalOutcome>((resolve) => {
const interactionId = newInteractionId()
const pending: PendingApproval = {
interactionId,
sessionId: request.agent.session.id,
approvalId,
toolName: request.toolName,
...(request.callId === undefined ? {} : { callId: request.callId }),
...(request.reason === undefined ? {} : { reason: request.reason }),
...(request.signal === undefined ? {} : { signal: request.signal }),
settle: (outcome) => {
if (this.approvals.get(interactionId) !== pending) return
this.approvals.delete(interactionId)
request.signal?.removeEventListener('abort', onAbort)
this.broadcast({
type: 'approval/resolved',
interactionId,
sessionId: pending.sessionId,
approvalId,
outcome,
})
resolve(outcome)
},
}
const onAbort = (): void => { pending.settle('cancelled') }
Object.assign(pending, { onAbort })
this.approvals.set(interactionId, pending)
request.signal?.addEventListener('abort', onAbort, { once: true })
if (request.signal?.aborted === true) {
pending.settle('cancelled')
return
}
this.broadcast({ type: 'approval/requested', ...approvalRequest(pending) })
})
}
private respondApproval(
pending: PendingApproval,
request: SessionRespondRequest,
): SessionRespondReceipt {
if (!request.result.ok) return { accepted: false, reason: 'bad-response' }
const response = approvalResponse(request.result.value)
if (response === undefined
|| response.sessionId !== pending.sessionId
|| response.approvalId !== pending.approvalId) {
return { accepted: false, reason: 'bad-response' }
}
pending.settle(response.outcome)
return { accepted: true }
}
private respondQuestion(
pending: PendingQuestion,
request: SessionRespondRequest,
): SessionRespondReceipt {
if (!request.result.ok) {
if (request.result.error.code !== 'cancelled') return { accepted: false, reason: 'bad-response' }
if (!this.claimQuestion(pending, 'cancelled')) return { accepted: false, reason: 'not-pending' }
pending.reject(new UserQuestionError(
'the user cancelled ask_user_question',
'ASK_CANCELLED',
))
return { accepted: true }
}
const response = questionResponse(request.result.value)
if (response === undefined
|| response.sessionId !== pending.sessionId
|| !matchesQuestions(response.answer, pending.questions)) {
return { accepted: false, reason: 'bad-response' }
}
if (!this.claimQuestion(pending, 'answered')) return { accepted: false, reason: 'not-pending' }
pending.resolve(response.answer)
return { accepted: true }
}
private claimQuestion(
pending: PendingQuestion,
outcome: 'answered' | 'cancelled',
): boolean {
if (this.questions.get(pending.interactionId) !== pending) return false
this.questions.delete(pending.interactionId)
if (pending.signal !== undefined && pending.onAbort !== undefined) {
pending.signal.removeEventListener('abort', pending.onAbort)
}
this.broadcast({
type: 'question/resolved',
interactionId: pending.interactionId,
sessionId: pending.sessionId,
outcome,
})
return true
}
private cancelSessionInteractions(sessionId: SessionId): void {
for (const pending of [...this.approvals.values()]) {
if (pending.sessionId === sessionId) pending.settle('cancelled')
}
for (const pending of [...this.questions.values()]) {
if (pending.sessionId !== sessionId || !this.claimQuestion(pending, 'cancelled')) continue
pending.reject(new UserQuestionError(
'the owning session was disposed before the user answered',
'ASK_ABORTED',
))
}
}
private broadcast(frame: SessionControlFrame): void {
for (const stream of this.streams) stream.push(frame)
}
@@ -395,29 +174,6 @@ class ControlQueue {
}
}
function newInteractionId(): SessionInteractionId {
return randomUUID() as SessionInteractionId
}
function approvalRequest(pending: PendingApproval): SessionApprovalRequest {
return {
interactionId: pending.interactionId,
sessionId: pending.sessionId,
approvalId: pending.approvalId,
toolName: pending.toolName,
...(pending.callId === undefined ? {} : { callId: pending.callId }),
...(pending.reason === undefined ? {} : { reason: pending.reason }),
}
}
function questionRequest(pending: PendingQuestion): SessionQuestionRequest {
return {
interactionId: pending.interactionId,
sessionId: pending.sessionId,
questions: pending.questions,
}
}
function queueItems(
agent: Agent,
splice?: SessionEventMap['agent/inbox/spliced'],
@@ -453,66 +209,3 @@ function jobView(job: JobSnapshot): SessionJob {
...(job.finishedAt === undefined ? {} : { finishedAt: job.finishedAt }),
}
}
function findApprovalId(
request: ApprovalRequestEvent,
pending: Iterable<PendingApproval>,
): ApprovalRequestId | undefined {
const claimed = new Set<ApprovalRequestId>()
for (const entry of pending) claimed.add(entry.approvalId)
const decided = new Set<ApprovalRequestId>()
const events = request.agent.session.events
for (let index = events.length - 1; index >= 0; index--) {
const event = events[index] as SessionEvent
if (event.type === 'approval/decided') {
decided.add(event.data.id)
continue
}
if (event.type !== 'approval/asked'
|| decided.has(event.data.id)
|| claimed.has(event.data.id)
|| (request.callId ?? null) !== (event.data.callId ?? null)) continue
return event.data.id
}
return undefined
}
function approvalResponse(value: unknown): SessionApprovalResponse | undefined {
if (!isRecord(value)
|| typeof value.sessionId !== 'string'
|| typeof value.approvalId !== 'string'
|| (value.outcome !== 'allowed-once' && value.outcome !== 'rejected')) return undefined
return value as unknown as SessionApprovalResponse
}
function questionResponse(value: unknown): SessionQuestionResponse | undefined {
if (!isRecord(value) || typeof value.sessionId !== 'string' || !isRecord(value.answer)) return undefined
const answers = value.answer.answers
if (!Array.isArray(answers) || !answers.every(answer => isRecord(answer)
&& typeof answer.id === 'string'
&& Array.isArray(answer.selected)
&& answer.selected.every(item => typeof item === 'string')
&& (answer.custom === undefined || typeof answer.custom === 'string'))) return undefined
return value as unknown as SessionQuestionResponse
}
function matchesQuestions(
answer: AskUserQuestionAnswer,
questions: readonly AskUserQuestionItem[],
): boolean {
if (answer.answers.length !== questions.length) return false
return answer.answers.every((item, index) => {
const question = questions[index] as AskUserQuestionItem
if (item.id !== question.id || new Set(item.selected).size !== item.selected.length) return false
const custom = item.custom?.trim()
if (custom !== undefined && custom === '') return false
if (question.multiSelect !== true
&& (item.selected.length > 1 || (custom !== undefined && item.selected.length > 0))) return false
const labels = new Set(question.options?.map(option => option.label) ?? [])
return item.selected.every(label => labels.has(label))
})
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
}
@@ -36,8 +36,6 @@ import type {
SessionPromptValue,
SessionRenameRequest,
SessionRenameValue,
SessionRespondReceipt,
SessionRespondRequest,
SessionSearchRequest,
SessionSearchValue,
SessionSelectModelRequest,
@@ -73,7 +71,6 @@ export class SessionController extends TypertRemoteService {
'sessionQuery',
'tools',
'typert',
'userQuestions',
'workspaceRegistry',
]
@@ -292,15 +289,6 @@ export class SessionController extends TypertRemoteService {
return this.controlState.control(signal)
}
/**
* Settle one still-pending approval or structured question.
* @param request - interaction identity and caller response.
* @returns whether a matching pending interaction accepted the response.
*/
@Remote('respond')
respond(request: SessionRespondRequest): SessionRespondReceipt {
return this.controlState.respond(request)
}
}
export { buildModelCatalog } from './catalog.ts'
+1 -76
View File
@@ -4,13 +4,11 @@ import type {
AttachmentIdType, ImageAttachmentLimits, ImageAttachmentRef, ImageMediaType,
} from '@deepseek-ai/dsh-attachment'
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { CallId, MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { JsonValue, SessionId, SurfaceOp } from '@deepseek-ai/dsh-session/types'
import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
import type { ApprovalOutcome, ApprovalRequestId } from '@deepseek-ai/dsh-user-approval/types'
import type { AskUserQuestionAnswer, AskUserQuestionItem } from '@deepseek-ai/dsh-user-questions/types'
import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types'
import type {
DiffCallView,
@@ -412,35 +410,13 @@ export interface SessionJob {
readonly finishedAt?: number
}
/** Stable identity of one answerable Host interaction. */
export type SessionInteractionId = Branded<'session-interaction-id'>
/** Complete live control baseline emitted once per control stream generation. */
export interface SessionControlBaseline {
readonly queues: Readonly<Record<SessionId, readonly SessionQueuedItem[]>>
readonly jobs: Readonly<Record<SessionId, readonly SessionJob[]>>
readonly approvals: readonly SessionApprovalRequest[]
readonly questions: readonly SessionQuestionRequest[]
readonly projections: Readonly<Record<SessionId, SessionProjectionsBlock>>
}
/** One pending approval request. */
export interface SessionApprovalRequest {
readonly interactionId: SessionInteractionId
readonly sessionId: SessionId
readonly approvalId: ApprovalRequestId
readonly toolName: string
readonly callId?: CallId
readonly reason?: string
}
/** One pending structured-question request. */
export interface SessionQuestionRequest {
readonly interactionId: SessionInteractionId
readonly sessionId: SessionId
readonly questions: readonly AskUserQuestionItem[]
}
/** One finished projection value and its durable watermark. */
export interface SessionProjectionUpdate {
readonly sessionId: SessionId
@@ -454,59 +430,8 @@ export type SessionControlFrame =
| { readonly type: 'baseline'; readonly value: SessionControlBaseline }
| { readonly type: 'queue'; readonly sessionId: SessionId; readonly items: readonly SessionQueuedItem[] }
| { readonly type: 'jobs'; readonly sessionId: SessionId; readonly jobs: readonly SessionJob[] }
| ({ readonly type: 'approval/requested' } & SessionApprovalRequest)
| {
readonly type: 'approval/resolved'
readonly interactionId: SessionInteractionId
readonly sessionId: SessionId
readonly approvalId: ApprovalRequestId
readonly outcome: ApprovalOutcome
}
| ({ readonly type: 'question/requested' } & SessionQuestionRequest)
| {
readonly type: 'question/resolved'
readonly interactionId: SessionInteractionId
readonly sessionId: SessionId
readonly outcome: 'answered' | 'cancelled'
}
| ({ readonly type: 'projection' } & SessionProjectionUpdate)
/** Result shell sent back for one pending interaction. */
export type SessionInteractionResult =
| { readonly ok: true; readonly value: SessionApprovalResponse | SessionQuestionResponse }
| {
readonly ok: false
readonly error: {
readonly code: string
readonly message: string
readonly details: Readonly<Record<string, JsonValue>>
}
}
/** Pending-interaction response request. */
export interface SessionRespondRequest {
readonly interactionId: SessionInteractionId
readonly result: SessionInteractionResult
}
/** Receipt for one pending-interaction response. */
export type SessionRespondReceipt =
| { readonly accepted: true }
| { readonly accepted: false; readonly reason: 'not-pending' | 'bad-response' }
/** Validated approval answer carried inside a successful interaction response. */
export interface SessionApprovalResponse {
readonly sessionId: SessionId
readonly approvalId: ApprovalRequestId
readonly outcome: 'allowed-once' | 'rejected'
}
/** Validated structured-question answer carried inside a successful interaction response. */
export interface SessionQuestionResponse {
readonly sessionId: SessionId
readonly answer: AskUserQuestionAnswer
}
declare module '@deepseek-ai/cordis' {
interface Events {
/**
@@ -1,450 +0,0 @@
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SessionStore from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ApprovalService from '@deepseek-ai/dsh-user-approval'
import type { ApprovalRequestId } from '@deepseek-ai/dsh-user-approval'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { describe, expect, it, vi } from 'vitest'
import { SessionControlController } from '../src/control.ts'
import type {
SessionApprovalRequest,
SessionControlFrame,
SessionInteractionId,
SessionRespondRequest,
} from '../src/types.ts'
interface ControlCapture {
readonly frames: SessionControlFrame[]
waitFor(type: SessionControlFrame['type']): Promise<SessionControlFrame>
}
async function harness(): Promise<{ ctx: Context; control: SessionControlController }> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
await ctx.plugin(ApprovalService)
const control = new SessionControlController(ctx)
await new Promise(resolve => setTimeout(resolve, 0))
return { ctx, control }
}
function agentOf(ctx: Context): Agent {
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1 })
return { session } as unknown as Agent
}
function openControl(control: SessionControlController, abort: AbortController): ControlCapture {
const frames: SessionControlFrame[] = []
const waiters: {
type: SessionControlFrame['type']
resolve(frame: SessionControlFrame): void
}[] = []
void (async () => {
for await (const frame of control.control(abort.signal)) {
frames.push(frame)
for (let index = waiters.length - 1; index >= 0; index--) {
const waiter = waiters[index] as (typeof waiters)[number]
if (waiter.type !== frame.type) continue
waiters.splice(index, 1)
waiter.resolve(frame)
}
}
})()
return {
frames,
waitFor: (type) => {
const found = frames.find(frame => frame.type === type)
if (found !== undefined) return Promise.resolve(found)
return new Promise((resolve) => { waiters.push({ type, resolve }) })
},
}
}
function requestedOf(frame: SessionControlFrame): SessionApprovalRequest {
if (frame.type !== 'approval/requested') {
throw new Error(`expected approval/requested, got ${frame.type}`)
}
return frame
}
async function waitForCount(
stream: ControlCapture,
type: SessionControlFrame['type'],
count: number,
): Promise<void> {
for (let index = 0; index < 200 && stream.frames.filter(frame => frame.type === type).length < count; index++) {
await new Promise(resolve => setTimeout(resolve, 5))
}
expect(stream.frames.filter(frame => frame.type === type).length).toBeGreaterThanOrEqual(count)
}
function answer(
interactionId: SessionInteractionId,
sessionId: unknown,
approvalId: ApprovalRequestId,
outcome: 'allowed-once' | 'rejected',
): SessionRespondRequest {
return {
interactionId,
result: { ok: true, value: { sessionId, approvalId, outcome } },
} as SessionRespondRequest
}
describe('approval pending registry', () => {
it('round-trips ask through requested, response, outcome, and resolved frames', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const agent = agentOf(ctx)
const asked = ctx.approval.request({ agent, toolName: 'bash', reason: 'sandbox escalation' })
const requested = requestedOf(await stream.waitFor('approval/requested'))
expect(requested).toMatchObject({
toolName: 'bash',
reason: 'sandbox escalation',
sessionId: agent.session.id,
})
expect(control.respond(answer(
requested.interactionId,
requested.sessionId,
requested.approvalId,
'allowed-once',
))).toEqual({ accepted: true })
await expect(asked).resolves.toBe('allowed-once')
const resolved = await stream.waitFor('approval/resolved')
expect(resolved).toMatchObject({ approvalId: requested.approvalId, outcome: 'allowed-once' })
expect(control.respond(answer(
requested.interactionId,
requested.sessionId,
requested.approvalId,
'rejected',
))).toEqual({ accepted: false, reason: 'not-pending' })
abort.abort()
})
it('replays one pending request with the same interaction id in a new baseline', async () => {
const { ctx, control } = await harness()
const firstAbort = new AbortController()
const first = openControl(control, firstAbort)
const agent = agentOf(ctx)
const asked = ctx.approval.request({ agent, toolName: 'write' })
const requested = requestedOf(await first.waitFor('approval/requested'))
firstAbort.abort()
const secondAbort = new AbortController()
const second = openControl(control, secondAbort)
const baseline = await second.waitFor('baseline')
if (baseline.type !== 'baseline') throw new Error('expected baseline')
const replayed = baseline.value.approvals[0]
expect(replayed?.interactionId).toBe(requested.interactionId)
expect(replayed?.approvalId).toBe(requested.approvalId)
expect(control.respond(answer(
requested.interactionId,
requested.sessionId,
requested.approvalId,
'rejected',
))).toEqual({ accepted: true })
await expect(asked).resolves.toBe('rejected')
secondAbort.abort()
})
it('rejects malformed and mismatched answers, and reports unknown interactions', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const agent = agentOf(ctx)
void ctx.approval.request({ agent, toolName: 'bash' })
const requested = requestedOf(await stream.waitFor('approval/requested'))
expect(control.respond(answer(
'ghost' as SessionInteractionId,
requested.sessionId,
requested.approvalId,
'rejected',
))).toEqual({ accepted: false, reason: 'not-pending' })
expect(control.respond({
interactionId: requested.interactionId,
result: { ok: false, error: { code: 'internal', message: 'x', details: {} } },
})).toEqual({ accepted: false, reason: 'bad-response' })
expect(control.respond(answer(
requested.interactionId,
requested.sessionId,
'other-approval' as ApprovalRequestId,
'rejected',
))).toEqual({ accepted: false, reason: 'bad-response' })
expect(control.respond({
interactionId: requested.interactionId,
result: { ok: true, value: { nonsense: 1 } as never },
})).toEqual({ accepted: false, reason: 'bad-response' })
abort.abort()
})
it('withdraws an approval when its ask signal aborts', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const agent = agentOf(ctx)
const cancel = new AbortController()
const asked = ctx.approval.request({ agent, toolName: 'bash', signal: cancel.signal })
const requested = requestedOf(await stream.waitFor('approval/requested'))
cancel.abort()
await expect(asked).resolves.toBe('cancelled')
expect(await stream.waitFor('approval/resolved')).toMatchObject({
approvalId: requested.approvalId,
outcome: 'cancelled',
})
expect(control.respond(answer(
requested.interactionId,
requested.sessionId,
requested.approvalId,
'allowed-once',
))).toEqual({ accepted: false, reason: 'not-pending' })
abort.abort()
})
it('settles a pre-aborted dispatch without publishing it', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1 })
session.append('approval/asked', {
id: 'pre-aborted' as ApprovalRequestId,
toolName: 'bash',
})
const agent = { session } as unknown as Agent
const cancelled = new AbortController()
cancelled.abort()
const outcome = await ctx.waterfall(
'approval/request',
{ agent, toolName: 'bash', signal: cancelled.signal },
() => Promise.resolve('unavailable' as const),
)
expect(outcome).toBe('cancelled')
const secondAbort = new AbortController()
const second = openControl(control, secondAbort)
await second.waitFor('baseline')
expect(second.frames.some(frame => frame.type === 'approval/requested')).toBe(false)
secondAbort.abort()
abort.abort()
void stream
})
it('settles pending approvals when the controller is disposed', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
await ctx.plugin(ApprovalService)
let control!: SessionControlController
const fiber = ctx.plugin(Object.assign((fiberCtx: Context) => {
control = new SessionControlController(fiberCtx)
}, { inject: ['sessions', 'agents', 'userQuestions', 'approval'] }))
await fiber.await()
const abort = new AbortController()
const stream = openControl(control, abort)
const asked = ctx.approval.request({ agent: agentOf(ctx), toolName: 'bash' })
const requested = requestedOf(await stream.waitFor('approval/requested'))
await fiber.dispose()
await expect(asked).resolves.toBe('cancelled')
expect(await stream.waitFor('approval/resolved')).toMatchObject({
approvalId: requested.approvalId,
outcome: 'cancelled',
})
abort.abort()
})
it('carries callId and ignores an abort after the answer settled', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const agent = agentOf(ctx)
const cancel = new AbortController()
vi.spyOn(cancel.signal, 'removeEventListener').mockImplementation(() => {})
const asked = ctx.approval.request({
agent,
toolName: 'bash',
callId: 'call-9' as never,
signal: cancel.signal,
})
const requested = requestedOf(await stream.waitFor('approval/requested'))
expect(requested.callId).toBe('call-9')
expect(control.respond(answer(
requested.interactionId,
requested.sessionId,
requested.approvalId,
'allowed-once',
))).toEqual({ accepted: true })
await expect(asked).resolves.toBe('allowed-once')
cancel.abort()
expect(stream.frames.filter(frame => frame.type === 'approval/resolved')).toHaveLength(1)
abort.abort()
})
it('contains an abort that wins immediately after pending registration', async () => {
const { ctx, control } = await harness()
void control
let reads = 0
const signal = {
get aborted() { return ++reads >= 3 },
addEventListener: () => {},
removeEventListener: () => {},
} as unknown as AbortSignal
await expect(ctx.approval.request({
agent: agentOf(ctx),
toolName: 'bash',
signal,
})).resolves.toBe('cancelled')
})
it('cancels only approvals owned by a disposed Session', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const first = agentOf(ctx)
const second = agentOf(ctx)
const firstAsk = ctx.approval.request({ agent: first, toolName: 'first' })
const secondAsk = ctx.approval.request({ agent: second, toolName: 'second' })
await waitForCount(stream, 'approval/requested', 2)
const requests = stream.frames.filter(
(frame): frame is Extract<SessionControlFrame, { type: 'approval/requested' }> => (
frame.type === 'approval/requested'
),
)
ctx.emit('session/disposed', first.session)
await expect(firstAsk).resolves.toBe('cancelled')
const remaining = requests.find(request => request.sessionId === second.session.id)
if (remaining === undefined) throw new Error('missing second approval')
expect(control.respond(answer(
remaining.interactionId,
remaining.sessionId,
remaining.approvalId,
'allowed-once',
))).toEqual({ accepted: true })
await expect(secondAsk).resolves.toBe('allowed-once')
abort.abort()
})
it('pairs parallel asks by callId', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const agent = agentOf(ctx)
const askA = ctx.approval.request({ agent, toolName: 'bash', callId: 'call-a' as never })
const askB = ctx.approval.request({ agent, toolName: 'bash', callId: 'call-b' as never })
await waitForCount(stream, 'approval/requested', 2)
const requests = stream.frames
.filter((frame): frame is Extract<SessionControlFrame, { type: 'approval/requested' }> => (
frame.type === 'approval/requested'
))
const requestA = requests.find(frame => frame.callId === 'call-a')
const requestB = requests.find(frame => frame.callId === 'call-b')
if (requestA === undefined || requestB === undefined) throw new Error('missing parallel request')
const askedIdByCall = new Map(agent.session.events
.filter(event => event.type === 'approval/asked')
.map(event => [String(event.data.callId), event.data.id]))
expect(requestA.approvalId).toBe(askedIdByCall.get('call-a'))
expect(requestB.approvalId).toBe(askedIdByCall.get('call-b'))
expect(control.respond(answer(
requestB.interactionId,
requestB.sessionId,
requestB.approvalId,
'rejected',
))).toEqual({ accepted: true })
expect(control.respond(answer(
requestA.interactionId,
requestA.sessionId,
requestA.approvalId,
'allowed-once',
))).toEqual({ accepted: true })
await expect(askA).resolves.toBe('allowed-once')
await expect(askB).resolves.toBe('rejected')
abort.abort()
})
it('gives parallel callId-less asks distinct audit ids', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const agent = agentOf(ctx)
const askA = ctx.approval.request({ agent, toolName: 'alpha' })
const askB = ctx.approval.request({ agent, toolName: 'beta' })
await waitForCount(stream, 'approval/requested', 2)
const requests = stream.frames
.filter((frame): frame is Extract<SessionControlFrame, { type: 'approval/requested' }> => (
frame.type === 'approval/requested'
))
const requestA = requests.find(frame => frame.toolName === 'alpha')
const requestB = requests.find(frame => frame.toolName === 'beta')
if (requestA === undefined || requestB === undefined) throw new Error('missing parallel request')
expect(requestA.approvalId).not.toBe(requestB.approvalId)
expect(control.respond(answer(
requestA.interactionId,
requestA.sessionId,
requestA.approvalId,
'allowed-once',
))).toEqual({ accepted: true })
expect(control.respond(answer(
requestB.interactionId,
requestB.sessionId,
requestB.approvalId,
'rejected',
))).toEqual({ accepted: true })
await expect(askA).resolves.toBe('allowed-once')
await expect(askB).resolves.toBe('rejected')
abort.abort()
})
it('delegates a dispatch whose only asked candidate is already decided', async () => {
const { ctx, control } = await harness()
void control
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1 })
session.append('approval/asked', {
id: 'stale-ask' as ApprovalRequestId,
toolName: 'bash',
})
session.append('approval/decided', {
id: 'stale-ask' as ApprovalRequestId,
outcome: 'rejected',
})
const agent = { session } as unknown as Agent
const outcome = await ctx.waterfall(
'approval/request',
{ agent, toolName: 'bash' },
() => Promise.resolve('unavailable' as const),
)
expect(outcome).toBe('unavailable')
})
it('delegates an ask with no matching audit event', async () => {
const { ctx, control } = await harness()
void control
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1 })
const agent = { session } as unknown as Agent
const outcome = await ctx.waterfall(
'approval/request',
{ agent, toolName: 'x' },
() => Promise.resolve('unavailable' as const),
)
expect(outcome).toBe('unavailable')
})
})
@@ -5,7 +5,6 @@ import type { JobOutcome } from '@deepseek-ai/dsh-jobs'
import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { describe, expect, it } from 'vitest'
import { SessionControlController } from '../src/control.ts'
import type { SessionControlFrame } from '../src/types.ts'
@@ -36,7 +35,6 @@ async function harness(withRegistry: boolean): Promise<{
}> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
if (withRegistry) {
await ctx.plugin(LocalJobRegistry)
@@ -1,354 +0,0 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from '@deepseek-ai/cordis'
import AgentRegistry, { Inbox, type Agent } from '@deepseek-ai/dsh-agent'
import SessionStore from '@deepseek-ai/dsh-session'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { SessionControlController } from '../src/control.ts'
import type {
SessionControlFrame,
SessionQuestionRequest,
SessionRespondRequest,
} from '../src/types.ts'
type QuestionFrame = Extract<SessionControlFrame, { type: 'question/requested' }>
async function harness(): Promise<{ ctx: Context; control: SessionControlController }> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
return { ctx, control: new SessionControlController(ctx) }
}
function agent(ctx: Context): Agent {
const session = ctx.sessions.create()
const inbox = new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
const value = { id: session.id, session, inbox, status: 'idle', ctx } as Agent
ctx.agents.register(value)
return value
}
function openControl(control: SessionControlController, abort: AbortController): {
frames: SessionControlFrame[]
waitForQuestion(): Promise<QuestionFrame>
} {
const frames: SessionControlFrame[] = []
let resolveQuestion!: (value: QuestionFrame) => void
const question = new Promise<QuestionFrame>((resolve) => {
resolveQuestion = resolve
})
void (async () => {
for await (const frame of control.control(abort.signal)) {
frames.push(frame)
if (frame.type === 'question/requested') resolveQuestion(frame)
}
})()
return { frames, waitForQuestion: () => question }
}
function answer(
request: SessionQuestionRequest,
selected: string[],
custom?: string,
): SessionRespondRequest {
const question = request.questions[0]
if (question === undefined) throw new Error('question request is empty')
return {
interactionId: request.interactionId,
result: {
ok: true,
value: {
sessionId: request.sessionId,
answer: {
answers: [{
id: question.id,
selected,
...custom === undefined ? {} : { custom },
}],
},
},
},
}
}
describe('question response validation', () => {
it('rejects questions without an owning Agent', async () => {
const { ctx } = await harness()
await expect(ctx.userQuestions.ask({
questions: [{ id: 'owner', question: 'Who owns this?', options: [{ label: 'Nobody' }] }],
})).rejects.toMatchObject({ code: 'ASK_MISSING_AGENT' })
})
it('accepts selected options with custom text for multi-select questions', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const asked = ctx.userQuestions.ask({
agent: agent(ctx),
questions: [{
id: 'targets',
question: 'Choose targets and add another',
multiSelect: true,
options: [{ label: 'Code' }, { label: 'Docs' }],
}],
})
const request = await stream.waitForQuestion()
expect(control.respond(answer(request, ['Code', 'Docs'], 'Release notes')))
.toEqual({ accepted: true })
await expect(asked).resolves.toEqual({
answers: [{ id: 'targets', selected: ['Code', 'Docs'], custom: 'Release notes' }],
})
expect(stream.frames.some(item => item.type === 'question/resolved')).toBe(true)
abort.abort()
})
it('keeps selected options and custom text mutually exclusive for single-select questions', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const asked = ctx.userQuestions.ask({
agent: agent(ctx),
questions: [{
id: 'target',
question: 'Choose one target',
options: [{ label: 'Code' }, { label: 'Docs' }],
}],
})
const request = await stream.waitForQuestion()
expect(control.respond(answer(request, ['Code'], 'Release notes')))
.toEqual({ accepted: false, reason: 'bad-response' })
expect(control.respond(answer(request, [], 'Release notes')))
.toEqual({ accepted: true })
await expect(asked).resolves.toEqual({
answers: [{ id: 'target', selected: [], custom: 'Release notes' }],
})
abort.abort()
})
it('rejects malformed answers without consuming the pending question', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const asked = ctx.userQuestions.ask({
agent: agent(ctx),
questions: [{
id: 'target',
question: 'Choose targets',
multiSelect: true,
options: [{ label: 'Code' }, { label: 'Docs' }],
}],
})
const request = await stream.waitForQuestion()
const malformed: unknown[] = [
null,
[],
{ sessionId: request.sessionId, answer: null },
{ sessionId: request.sessionId, answer: { answers: 'invalid' } },
{ sessionId: request.sessionId, answer: { answers: [null] } },
{ sessionId: request.sessionId, answer: { answers: [{ id: 'target', selected: [1] }] } },
{ sessionId: request.sessionId, answer: { answers: [{ id: 'target', selected: [], custom: 1 }] } },
{ sessionId: 'other', answer: { answers: [{ id: 'target', selected: [] }] } },
{ sessionId: request.sessionId, answer: { answers: [] } },
{ sessionId: request.sessionId, answer: { answers: [{ id: 'other', selected: [] }] } },
{ sessionId: request.sessionId, answer: { answers: [{ id: 'target', selected: ['Code', 'Code'] }] } },
{ sessionId: request.sessionId, answer: { answers: [{ id: 'target', selected: [], custom: ' ' }] } },
{ sessionId: request.sessionId, answer: { answers: [{ id: 'target', selected: ['Unknown'] }] } },
]
expect(control.respond({
interactionId: request.interactionId,
result: { ok: false, error: { code: 'internal', message: 'bad', details: {} } },
})).toEqual({ accepted: false, reason: 'bad-response' })
for (const value of malformed) {
expect(control.respond({
interactionId: request.interactionId,
result: { ok: true, value: value as never },
})).toEqual({ accepted: false, reason: 'bad-response' })
}
expect(control.respond(answer(request, ['Code']))).toEqual({ accepted: true })
await expect(asked).resolves.toEqual({ answers: [{ id: 'target', selected: ['Code'] }] })
abort.abort()
})
it('accepts free-form answers when a question has no options', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const asked = ctx.userQuestions.ask({
agent: agent(ctx),
questions: [{ id: 'detail', question: 'Provide detail' }],
})
const request = await stream.waitForQuestion()
expect(control.respond(answer(request, [], 'details'))).toEqual({ accepted: true })
await expect(asked).resolves.toEqual({
answers: [{ id: 'detail', selected: [], custom: 'details' }],
})
abort.abort()
})
it('handles caller cancellation and races while a response is decoded', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const cancelledAsk = ctx.userQuestions.ask({
agent: agent(ctx),
questions: [{ id: 'cancel', question: 'Cancel?', options: [{ label: 'No' }] }],
})
const cancelled = await stream.waitForQuestion()
expect(control.respond({
interactionId: cancelled.interactionId,
result: { ok: false, error: { code: 'cancelled', message: 'cancelled', details: {} } },
})).toEqual({ accepted: true })
await expect(cancelledAsk).rejects.toMatchObject({ code: 'ASK_CANCELLED' })
const raceAbort = new AbortController()
const racedAsk = ctx.userQuestions.ask({
agent: agent(ctx),
signal: raceAbort.signal,
questions: [{ id: 'race', question: 'Race?', options: [{ label: 'Yes' }] }],
})
const raced = await vi.waitFor(() => {
const found = stream.frames.find(frame => frame.type === 'question/requested'
&& frame.questions[0]?.id === 'race')
expect(found).toBeDefined()
return found as QuestionFrame
})
const error = {
get code(): string {
raceAbort.abort()
return 'cancelled'
},
message: 'cancelled',
details: {},
}
expect(control.respond({
interactionId: raced.interactionId,
result: { ok: false, error },
})).toEqual({ accepted: false, reason: 'not-pending' })
await expect(racedAsk).rejects.toMatchObject({ code: 'ASK_ABORTED' })
const answerAbort = new AbortController()
const answerRace = ctx.userQuestions.ask({
agent: agent(ctx),
signal: answerAbort.signal,
questions: [{ id: 'answer-race', question: 'Race?', options: [{ label: 'Yes' }] }],
})
const answerRequest = await vi.waitFor(() => {
const found = stream.frames.find(frame => frame.type === 'question/requested'
&& frame.questions[0]?.id === 'answer-race')
expect(found).toBeDefined()
return found as QuestionFrame
})
const result = {
ok: true as const,
get value() {
answerAbort.abort()
return {
sessionId: answerRequest.sessionId,
answer: { answers: [{ id: 'answer-race', selected: ['Yes'] }] },
}
},
}
expect(control.respond({ interactionId: answerRequest.interactionId, result }))
.toEqual({ accepted: false, reason: 'not-pending' })
await expect(answerRace).rejects.toMatchObject({ code: 'ASK_ABORTED' })
abort.abort()
})
it('contains aborts before and immediately after provider registration', async () => {
const { ctx } = await harness()
const question = { id: 'race', question: 'Race?', options: [{ label: 'Yes' }] }
const signalAfter = (abortedAt: number, notifyOnAdd = false): AbortSignal => {
let reads = 0
return {
get aborted() { return ++reads >= abortedAt },
addEventListener: (_type: string, listener: EventListenerOrEventListenerObject) => {
if (!notifyOnAdd) return
if (typeof listener === 'function') listener(new Event('abort'))
else listener.handleEvent(new Event('abort'))
},
removeEventListener: () => {},
} as unknown as AbortSignal
}
await expect(ctx.userQuestions.ask({
agent: agent(ctx),
signal: signalAfter(2),
questions: [question],
})).rejects.toMatchObject({ code: 'ASK_ABORTED' })
await expect(ctx.userQuestions.ask({
agent: agent(ctx),
signal: signalAfter(3, true),
questions: [question],
})).rejects.toMatchObject({ code: 'ASK_ABORTED' })
})
it('replays pending questions in baselines and rejects them on controller disposal', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
let control!: SessionControlController
const fiber = ctx.plugin(Object.assign((fiberCtx: Context) => {
control = new SessionControlController(fiberCtx)
}, { inject: ['sessions', 'agents', 'userQuestions'] }))
await fiber.await()
const firstAbort = new AbortController()
const first = openControl(control, firstAbort)
const asked = ctx.userQuestions.ask({
agent: agent(ctx),
questions: [{ id: 'pending', question: 'Pending?', options: [{ label: 'Yes' }] }],
})
const requested = await first.waitForQuestion()
const secondAbort = new AbortController()
const second = openControl(control, secondAbort)
await vi.waitFor(() => { expect(second.frames[0]?.type).toBe('baseline') })
const baseline = second.frames[0]
if (baseline?.type !== 'baseline') throw new Error('missing baseline')
expect(baseline.value.questions).toContainEqual(expect.objectContaining({
interactionId: requested.interactionId,
}))
await fiber.dispose()
await expect(asked).rejects.toMatchObject({ code: 'ASK_ABORTED' })
firstAbort.abort()
secondAbort.abort()
})
it('cancels only questions owned by a disposed Session', async () => {
const { ctx, control } = await harness()
const abort = new AbortController()
const stream = openControl(control, abort)
const first = agent(ctx)
const second = agent(ctx)
const firstAsk = ctx.userQuestions.ask({
agent: first,
questions: [{ id: 'first', question: 'First?', options: [{ label: 'Yes' }] }],
})
const secondAsk = ctx.userQuestions.ask({
agent: second,
questions: [{ id: 'second', question: 'Second?', options: [{ label: 'Yes' }] }],
})
await vi.waitFor(() => {
expect(stream.frames.filter(frame => frame.type === 'question/requested')).toHaveLength(2)
})
const requests = stream.frames.filter(
(frame): frame is QuestionFrame => frame.type === 'question/requested',
)
ctx.emit('session/disposed', first.session)
await expect(firstAsk).rejects.toMatchObject({ code: 'ASK_ABORTED' })
const remaining = requests.find(request => request.sessionId === second.session.id)
if (remaining === undefined) throw new Error('missing second question')
expect(control.respond(answer(remaining, ['Yes']))).toEqual({ accepted: true })
await expect(secondAsk).resolves.toEqual({
answers: [{ id: 'second', selected: ['Yes'] }],
})
abort.abort()
})
})
@@ -3,7 +3,6 @@ import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { describe, expect, it } from 'vitest'
import { SessionControlController } from '../src/control.ts'
@@ -16,7 +15,6 @@ async function harness(): Promise<{
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const session = ctx.sessions.create(SessionId('queue-session'))
const inbox = new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} })
const agent = { id: session.id, session, inbox, status: 'running', ctx } as Agent
@@ -5,7 +5,6 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import { describe, expect, it, vi } from 'vitest'
import type { SessionInteractionId } from '../src/types.ts'
import { createSessionTestController } from './test-remote.ts'
const defaults = {
@@ -74,9 +73,5 @@ describe('SessionController facade', () => {
})
abort.abort()
await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined })
expect(controller.respond({
interactionId: 'missing' as SessionInteractionId,
result: { ok: false, error: { code: 'cancelled', message: 'cancelled', details: {} } },
})).toEqual({ accepted: false, reason: 'not-pending' })
})
})
@@ -16,7 +16,6 @@ import { TypertLookupFailure } from '@deepseek-ai/dsh-typert-protocol'
import TypertRegistry from '@deepseek-ai/dsh-typert-registry'
import { createUserMessage, MessageId } from '@deepseek-ai/dsh-llm'
import type { Agent } from '@deepseek-ai/dsh-agent'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionPromptRequest, SessionRequestId } from '../src/types.ts'
import {
@@ -51,7 +50,6 @@ describe('sessions.list cold merge', () => {
it('verifies only small possibly-blank artifacts and treats every unavailable probe as visible', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
const root = mkdtempSync(join(tmpdir(), 'dsh-cold-'))
const smallPath = join(root, 'small.log')
const largePath = join(root, 'large.log')
@@ -144,7 +142,6 @@ describe('sessions.list cold merge', () => {
it('can disable bounded blank probes without hiding cold Sessions', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
const meta = header('probe-disabled', 100)
const readFrom = vi.fn()
ctx.provide('sessionPersistence', {
@@ -169,7 +166,6 @@ describe('sessions.list cold merge', () => {
it('replaces a probed cold row with the live Session that attached during the read', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
const meta = header('attached-during-probe', 100)
const root = mkdtempSync(join(tmpdir(), 'dsh-cold-race-'))
@@ -227,7 +223,6 @@ describe('attached updatedAt tracks human prompts', () => {
it('ignores pickup and non-prompt work after the latest human message', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
@@ -277,7 +272,6 @@ describe('cold history recovery view', () => {
it('shows in-memory interruption repair without activating the session', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
const sessionId = sid('session-interrupted')
const meta = header(sessionId, 1000)
const stored: StoredPrefix<never> = {
@@ -344,7 +338,6 @@ describe('Remote Agent and Session lookup policy', () => {
await ctx.plugin(TypertRegistry)
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const sessionId = sid('session-remote-cold')
const meta = header(sessionId, 1000)
const inspect = vi.fn(() => Promise.resolve({ meta, events: [] as SessionEvent[] }))
@@ -386,7 +379,6 @@ describe('Remote Agent and Session lookup policy', () => {
await ctx.plugin(TypertRegistry)
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const coldId = sid('session-remote-cold-child')
const coldMeta = header(coldId, 1000, {
parentSession: sid('session-parent'),
@@ -437,7 +429,6 @@ describe('subagent ownership fence', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const sessionId = sid('session-child')
const meta = header('session-child', 1000, {
parentSession: sid('session-parent'),
@@ -507,7 +498,6 @@ describe('subagent ownership fence', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const sessionId = sid('session-legacy-child')
const meta = header('session-legacy-child', 1000, {
parentSession: sid('session-parent'),
@@ -548,7 +538,6 @@ describe('subagent ownership fence', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const parentSession = ctx.sessions.create(sid('session-parent'), { meta: { cwd: '/proj' } })
const parent = { id: parentSession.id, session: parentSession, status: 'idle', ctx } as Agent
ctx.agents.register(parent)
@@ -604,7 +593,6 @@ describe('subagent ownership fence', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const session = ctx.sessions.create(sid('session-ordinary-fork'), {
seed: [{
type: 'subagent/descriptor',
@@ -632,7 +620,6 @@ describe('subagent ownership fence', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const session = ctx.sessions.create(sid('session-browser-zone'), { meta: { cwd: '/proj' } })
const followup = vi.fn()
const agent = { id: session.id, session, status: 'idle', ctx, followup } as unknown as Agent
@@ -702,7 +689,6 @@ describe('degenerate composition (no persistence, no factory)', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
const listed = await remote.list(request({}))
@@ -725,7 +711,6 @@ describe('degenerate composition (no persistence, no factory)', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const inspect = vi.fn()
ctx.provide('sessionPersistence', {
list: () => Promise.resolve([]),
@@ -748,7 +733,6 @@ describe('sessions.prompt synchronous rejection', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const session = ctx.sessions.create(sid('session-throwing'))
// A live structural stub whose delivery verbs throw synchronously, the
// shape a disposed loop presents at this gateway boundary.
@@ -781,7 +765,6 @@ describe('sessions.prompt synchronous rejection', () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
const sessionId = sid('race-resume')
const meta: SessionHeader = header('race-resume', 1000)
ctx.provide('sessionPersistence', {
@@ -9,7 +9,6 @@ import type { LlmCallConfig } from '@deepseek-ai/dsh-llm'
import SessionStore from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import type { Workspace } from '@deepseek-ai/dsh-workspace'
import { createSessionTestRemote } from './test-remote.ts'
@@ -24,7 +23,6 @@ async function composed(workspaces: readonly Workspace[] = []): Promise<Context>
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
ctx.provide('workspaceRegistry', { list: () => workspaces } as never)
ctx.agents.setFactory({
createAgent: async (ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> => {
@@ -17,7 +17,6 @@ import { CallId, createMessage, createToolResultMessage, createUserMessage } fro
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import type { ToolDefinition } from '@deepseek-ai/dsh-tools'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
import type { SessionFollowFrame } from '@deepseek-ai/dsh-api-session-controller/types'
import { createSessionTestRemote } from './test-remote.ts'
@@ -68,7 +67,6 @@ async function harness(): Promise<{ ctx: Context }> {
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(ToolRuntime)
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
ctx.tools.register(tool('gen', {
presentCall: () => ({ card: 'generic', title: 'gen call' }),
@@ -13,18 +13,15 @@ import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SessionStore from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { CommandId } from '@deepseek-ai/dsh-commands/brand'
// Side-effect type imports: the knob-event SessionEventMap merges.
// Side-effect type imports: the configuration-event SessionEventMap merges.
import type {} from '@deepseek-ai/dsh-permission-presets'
import type {} from '@deepseek-ai/dsh-sandbox-policy'
import type {} from '@deepseek-ai/dsh-user-approval'
import { createSessionTestRemote, type TestSessionRemote } from './test-remote.ts'
async function harness(): Promise<{ ctx: Context; remote: TestSessionRemote; attach: (session: Session) => void }> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
return {
ctx,
@@ -45,10 +42,9 @@ function appendStandalone(session: Session): void {
session.append('session/title', {
title: 'standalone title', messageSeqs: [], source: { kind: 'fallback' },
})
// The three permission knob events (a /permission switch on a fresh session).
// Permission configuration events from a /permission switch on a fresh session.
session.append('permission/preset', { preset: 'danger-full-access' })
session.append('sandbox/mode', { mode: 'danger-full-access' })
session.append('approval/policy', { policy: 'never' })
}
async function listBlank(remote: TestSessionRemote, id: string): Promise<boolean | undefined> {
@@ -20,7 +20,6 @@ import SessionStore from '@deepseek-ai/dsh-session'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionPromptRequest, SessionRequestId } from '../src/types.ts'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { TypertRemoteFailure } from '@deepseek-ai/dsh-typert-protocol'
import { createSessionTestRemote } from './test-remote.ts'
@@ -95,7 +94,6 @@ async function harness(logged?: {
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, { persona: '' })
await ctx.plugin(LlmRuntime)
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
ctx.llm.registerAdapter(['deepseek-official'], new CatalogAdapter('DeepSeek', [
{ provider: 'deepseek-official', id: 'deepseek-chat', name: 'DeepSeek Chat' },
@@ -18,7 +18,6 @@ import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import { SessionControlController } from '@deepseek-ai/dsh-api-session-controller/src/control.ts'
import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller/types'
import { createSessionTestRemote, type TestSessionRemote } from './test-remote.ts'
@@ -76,7 +75,6 @@ const internalCountUnit = () => ({
async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: Session }> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserQuestionService)
await ctx.plugin(AgentRegistry)
if (withRegistry) await ctx.plugin(SessionProjectionRegistry)
const session = ctx.sessions.create()
@@ -274,7 +272,7 @@ describe('session.history projections block', () => {
expect('sessionListMetadata' in ctx.sessionProjections.snapshot(session).values).toBe(false)
const fiber = ctx.plugin(Object.assign((gatewayCtx: Context) => {
createSessionTestRemote(gatewayCtx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
}, { inject: ['sessions', 'agents', 'userQuestions', 'sessionProjections'] }))
}, { inject: ['sessions', 'agents', 'sessionProjections'] }))
await fiber.await()
await vi.waitFor(() => {
expect(ctx.sessionProjections.snapshot(session).values.sessionListMetadata)
@@ -14,7 +14,6 @@ import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent, AgentHandle, CreateAgentOptions } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionTitleService from '@deepseek-ai/dsh-session-title'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import type { Session, SessionId } from '@deepseek-ai/dsh-session'
import { createSessionTestRemote } from './test-remote.ts'
@@ -28,7 +27,6 @@ async function composed(withTitles = true): Promise<Context> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
if (withTitles) {
await ctx.plugin(SessionTitleService, { fallbackMaxWords: 5, fallbackMaxBytes: 40, maxTitleBytes: 40 })
}
@@ -11,7 +11,6 @@ import AgentRegistry from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore from '@deepseek-ai/dsh-session'
import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import UserQuestionService from '@deepseek-ai/dsh-user-questions'
import {
SessionQueryError,
type SessionSearchHit,
@@ -61,7 +60,6 @@ async function baseContext(): Promise<Context> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(AgentRegistry)
await ctx.plugin(UserQuestionService)
return ctx
}
@@ -93,12 +93,6 @@ function installControllers(
},
} as never)
}
if (ctx.get('userQuestions') === undefined) {
ctx.provide('userQuestions', {
registerProvider: () => (): void => {},
} as never)
}
const cwd = vi.spyOn(process, 'cwd').mockReturnValue(defaults.cwd)
let controller: SessionController
try {
@@ -15,8 +15,6 @@
{ "path": "../gateway/tsconfig.client.json" },
{ "path": "../../attachment/attachment" },
{ "path": "../../core/session" },
{ "path": "../../interaction/user-approval" },
{ "path": "../../interaction/user-questions" },
{ "path": "../../jobs/jobs" },
{ "path": "../../llm/llm" },
{ "path": "../../session/session-projection" },
@@ -26,8 +26,6 @@
{ "path": "../../core/session" },
{ "path": "../../core/tools" },
{ "path": "../../attachment/attachment" },
{ "path": "../../interaction/user-approval" },
{ "path": "../../interaction/user-questions" },
{ "path": "../../jobs/jobs" },
{ "path": "../../llm/llm" },
{ "path": "../../preset/agent-presets" },
-6
View File
@@ -1059,12 +1059,6 @@ importers:
'@deepseek-ai/dsh-typert-registry':
specifier: workspace:^
version: link:../../typert/registry
'@deepseek-ai/dsh-user-approval':
specifier: workspace:^
version: link:../../interaction/user-approval
'@deepseek-ai/dsh-user-questions':
specifier: workspace:^
version: link:../../interaction/user-questions
'@deepseek-ai/dsh-workspace':
specifier: workspace:^
version: link:../../workspace/workspace