/** Session Remote owner: cold reads, explicit Agent commands, and live control state. */ import { hostname } from 'node:os' import { resolve } from 'node:path' import type { Agent } from '@deepseek-ai/dsh-agent' import type {} from '@deepseek-ai/dsh-fs' import { Context, Service } from '@deepseek-ai/cordis' import z from '@deepseek-ai/schemastery' import { errorChain, ReasoningEffortId, type MessageSource } from '@deepseek-ai/dsh-llm' import type {} from '@deepseek-ai/dsh-client-file-upload' import { canOpenNativePath, nativeFileManager, nativeFileApplications, openNativeFileApplication, openNativeAssociatedPath, revealNativePath } from '@deepseek-ai/dsh-native-command' import type { SessionId } from '@deepseek-ai/dsh-session' import type { SessionInspection } from '@deepseek-ai/dsh-session-persistence' import type {} from '@deepseek-ai/dsh-session-projection-cache' import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query' import { Remote, RemoteError, TypertRemoteService, principalOf } from '@deepseek-ai/dsh-typert-protocol' import { canonicalClientTimeZone } from '@deepseek-ai/dsh-util-time' import { assertNever } from '@deepseek-ai/dsh-util-values' import type {} from '@deepseek-ai/dsh-api-remote-access' import { ApiSessionAgentController, inspectApiSession, type ApiSessionAgentResult, } from './agent.ts' import { SessionCommandController, hasPromptContent } from './commands.ts' import { SessionControlController } from './control.ts' import { SessionHistoryController } from './history.ts' import { SessionFileReferences } from './file-references.ts' import { ApiSessionList } from './list.ts' import { buildModelCatalog, hasProviderApiKey } from './catalog.ts' import { SessionSkillCatalog } from './skill-catalog.ts' import { SessionMediaReferences } from './media-references.ts' import { ArchivedSessionGate } from './archived-session-gate.ts' import type { ModelCatalog, SessionWorkspacePathApplication, SessionAddress, SessionAttachmentRequest, SessionAttachmentValue, SessionCancelRequest, SessionCancelValue, SessionControlFrame, SessionCreateRequest, SessionCreateValue, SessionFollowFrame, SessionFollowRequest, SessionForkRequest, SessionForkValue, SessionListRequest, SessionListValue, SessionOpenWorkspacePathRequest, SessionOpenWorkspacePathValue, SessionPage, SessionPageRequest, SessionPromptRequest, SessionPromptValue, SessionRenameRequest, SessionRenameValue, SessionSearchRequest, SessionSearchValue, SessionSelectModelRequest, SessionSelectModelValue, SessionProjectionsRequest, SessionProjectionsValue, SessionProjectionValues, SessionUpdateQueueRequest, SessionUpdateQueueValue, } from './types.ts' export type * from './types.ts' export { ApiSessionNotFound } from './agent.ts' export { SessionFileReferences } from './file-references.ts' export { SessionSkillCatalog } from './skill-catalog.ts' declare module '@deepseek-ai/cordis' { interface Context { /** Host Session business API and Remote namespace owner. */ sessionController: SessionController } } /** * Read the Session resource a history address names: the Session itself, or * the child Session whose transcript a subagent address reads. * @param request - history request carrying the durable address. * @returns the Session whose access the read requires. */ function addressedSession(request: { readonly address: SessionAddress }): SessionId { return request.address.kind === 'session' ? request.address.sessionId : request.address.childSessionId } /** Session Controller deployment policy. */ export interface Config { /** Override platform desktop-opener detection. */ readonly nativeOpen?: boolean /** Positive integral milliseconds of list work before yielding between complete rows. */ readonly listWorkSliceMs?: number } /** Deployment policy after schema defaults have been applied. */ type ResolvedConfig = Config & { readonly listWorkSliceMs: number } /** Host integrations replaceable by direct unit tests. */ export interface SessionControllerInternals { /** Native default-application handoff. */ readonly openPath?: (path: string, signal: AbortSignal) => Promise /** Native file-association query. */ readonly fileApplications?: typeof nativeFileApplications /** Explicit registered-application handoff. */ readonly openFileApplication?: typeof openNativeFileApplication /** Native file-manager handoff. */ readonly revealPath?: (path: string, signal: AbortSignal) => Promise /** Native handoff availability probe. */ readonly canOpenPath?: () => boolean } /** Host service backing the generated `ctx.remote.session` namespace. */ export class SessionController extends TypertRemoteService { static inject = [ 'agentDefaultModel', 'agents', 'attachments', 'fileUploads', 'fs', 'llm', 'remoteAccess', 'sessions', 'sessionProjections', 'sessionQuery', 'typert', 'workspaceRegistry', ] static Config: z = z.object({ nativeOpen: z.boolean(), listWorkSliceMs: z.natural().min(1).default(16), }) private agents!: ApiSessionAgentController private commands!: SessionCommandController private controlState!: SessionControlController private history!: SessionHistoryController private listState!: ApiSessionList private readonly openPath: (path: string, signal: AbortSignal) => Promise private readonly fileApplications: typeof nativeFileApplications private readonly openFileApplication: typeof openNativeFileApplication private readonly revealPath: (path: string, signal: AbortSignal) => Promise private readonly canOpenPath: () => boolean private readonly listWorkSliceMs: number private readonly promotions = new Set>() /** * @param ctx - Host context containing the Session capability assembly. * @param config - native-opener and list-scheduling deployment policy. * @param internals - host integrations replaceable by direct unit tests. */ constructor(ctx: Context, config: Config, internals: SessionControllerInternals = {}) { super(ctx, 'sessionController', { namespace: 'session' }) this.listWorkSliceMs = SessionController.Config(config).listWorkSliceMs this.fileApplications = internals.fileApplications ?? nativeFileApplications this.openFileApplication = internals.openFileApplication ?? openNativeFileApplication this.openPath = internals.openPath ?? openNativeAssociatedPath this.revealPath = internals.revealPath ?? revealNativePath this.canOpenPath = internals.canOpenPath ?? (() => config.nativeOpen ?? (internals.openPath !== undefined || canOpenNativePath())) } protected async [Service.init](): Promise { const ctx = this.ctx ctx.typert.resources.register('session', { field: 'sessionId' }) this.agents = new ApiSessionAgentController(ctx) await this.agents.initialize() this.commands = new SessionCommandController(ctx, this.agents, process.cwd()) ctx.effect(() => ctx.fileUploads.registerAgentResolver(async (sessionId) => { const result = await this.agents.resolveAgent(sessionId) if ('error' in result) throw result.error return result.agent }), 'session-controller: file-upload Agent resolver') this.controlState = new SessionControlController(ctx) // Registered before history so reverse-order teardown closes every // follower before waiting for already-admitted promotions. ctx.effect(() => async () => { await Promise.allSettled([...this.promotions]) }, 'session-controller.promotions') this.history = new SessionHistoryController(ctx, (observation) => { this.promote(observation) }) this.listState = new ApiSessionList(ctx, this.listWorkSliceMs) await this.listState.initialize() ctx.plugin(SessionFileReferences) ctx.plugin(SessionMediaReferences) ctx.plugin(SessionSkillCatalog) // An archived Session, or a subagent descendant of one, runs no model step // until it is restored; what it still runs is stopped by the owners that // answer the Workspace registry's archive-admission events. ctx.plugin(ArchivedSessionGate) ctx.on('session/created', (session) => { ctx.emit('api-session/added', this.listState.summaryFor(session)) }) ctx.on('session/disposed', (session) => { ctx.emit('api-session/removed', session.id) }) const publishAgentAvailability = ({ agent }: { agent: Agent }): undefined => { if (ctx.sessions.get(agent.id) === agent.session) { ctx.emit('api-session/added', this.listState.summaryFor(agent.session)) } } ctx.on('agent/created', publishAgentAvailability) ctx.on('agent/disposed', publishAgentAvailability) ctx.on('agent/status', ({ agent, status }) => { ctx.emit('api-session/status', agent.id, status === 'running') }) ctx.on('agent/error', ({ agent, error }) => { ctx.emit('api-session/error', agent.id, errorChain(error)) }) ctx.on('session/event', (session, event) => { if (event.type === 'request/header') { const agent = ctx.agents.get(session.id) if (agent?.session === session) this.agents.consumeSelection( agent, event.data.header.config.provider, event.data.header.config.model, event.data.header.config.reasoningEffort, ) } if (event.type !== 'user/message' || event.data.source.kind !== 'user') return ctx.emit('api-session/activity', session.id, event.time) }) } private promote(observation: SessionObservation): void { const sessionId = observation.header.id const task = (async () => { using ownedObservation = observation const result = await this.agents.resolveObservedAgent(ownedObservation) if ('error' in result) this.ctx.emit('api-session/error', sessionId, result.error.message) })().catch((error: unknown) => { this.ctx.logger.error(`session-controller: background activation for "${sessionId}" failed: ${errorChain(error)}`) }) this.promotions.add(task) void task.finally(() => { this.promotions.delete(task) }) } /** * Resolve or resume one ordinary Session for another Host API domain. * @param sessionId - Session identity whose Agent owns the operation. * @returns the live Agent or the stable Session-domain failure. */ resolveAgent(sessionId: SessionId): Promise { return this.agents.resolveAgent(sessionId) } /** * Inspect one attached or persisted Session without activating its Agent. * @param sessionId - durable Session identity. * @param signal - optional caller cancellation for persistence reads. * @returns the current attached state or persisted header and event prefix. */ inspect( sessionId: SessionId, signal?: AbortSignal, ): Promise { const attached = this.ctx.sessions.get(sessionId) if (attached !== undefined) { return Promise.resolve({ meta: attached.header, inheritedEventCount: attached.inheritedEventCount, // oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred. events: attached.snapshotEvents(), }) } return inspectApiSession(this.ctx, sessionId, signal) } /** * Read all visible Session rows without resuming an Agent. * @param _request - reserved empty list request. * @param signal - cancellation for persistence reads and summary generation. * @returns visible Session summaries ordered by activity. */ @Remote('list') async list(_request: SessionListRequest, signal: AbortSignal): Promise { const items = await this.listState.list(signal) return { items: items.filter(item => this.ctx.remoteAccess.allows({ kind: 'session', id: item.sessionId })) } } /** * Search visible Session content without resuming an Agent. A non-owner * receives only the hits whose Session `allows()` admits; `hasMore` * describes the bounded result before that filtering. * @param request - literal message-content query. * @param signal - cancellation for list and search reads. * @returns authorized bounded Session search results. */ @Remote('search') async search(request: SessionSearchRequest, signal: AbortSignal): Promise { const result = await this.listState.search(request.query, signal) return { ...result, items: result.items.filter(item => this.ctx.remoteAccess.allows({ kind: 'session', id: item.sessionId })) } } /** * Create one ordinary Session for a new conversation, reusing the named blank Session while it is reusable. * @param request - optional blank Session to reuse, location, and Agent preset. * @returns the created or reused Session identity and its preset when configured. */ @Remote('create') create(request: SessionCreateRequest): Promise { return this.commands.create(request) } /** * Select one Session-local model after explicitly resuming the Session; save the default in the background. * @param request - Session identity and requested model selection. * @returns the normalized selection installed for the Session, without waiting for default persistence. */ @Remote('selectModel') selectModel(request: SessionSelectModelRequest): Promise { return this.commands.selectModel(request) } /** * Select the first available account model after login when no provider API key is configured. * @returns after saving the first available model or retaining the existing default. */ @Remote async initializeDefaultModel(): Promise { const provider = 'deepseek-account' if (await hasProviderApiKey(this.ctx)) return const catalog = await buildModelCatalog(this.ctx) const model = catalog.groups.find(group => group.id === provider)?.models[0] if (model === undefined) throw new RemoteError('session/provider-models-unavailable', `provider "${provider}" has no available models`, { provider }) const selection = { provider, model: model.id, ...model.reasoning?.defaultEffort === undefined ? {} : { reasoningEffort: ReasoningEffortId(model.reasoning.defaultEffort) }, } await this.ctx.agentDefaultModel.saveSelection(selection) } /** * Describe every currently routable model for Host-generation selectors. * @returns provider-grouped models, the deployment default, and isolated provider failures. */ @Remote('modelCatalog') modelCatalog(): Promise { return buildModelCatalog(this.ctx) } /** * Report whether this deployment can hand a Session workspace path to a native desktop. * @returns true when the matching open operation is available. */ @Remote canOpenWorkspacePath(): boolean { return this.canOpenPath() } /** * Describe the serving desktop for authenticated file-action routes. * @returns Host name, configured availability, and platform-specific file-manager behavior. */ workspaceDesktop(): { name: string; available: boolean; fileManager: 'finder' | 'explorer' | 'directory' | null } { const fileManager = nativeFileManager() return { name: hostname(), available: fileManager !== null && this.canOpenPath(), fileManager } } /** * Verify one path through the composed filesystem and open it on the Host desktop. * @param request - path after best-effort Session workspace resolution. * @param signal - caller lifetime; abort terminates the native command. * @returns confirmation after the native opener accepts the path. * @throws RemoteError when the request is invalid, has no verified Host mapping, is cancelled, or the opener fails. */ @Remote('openWorkspacePath') async openWorkspacePath( request: SessionOpenWorkspacePathRequest, signal: AbortSignal, ): Promise { try { const path = await this.verifyDesktopPath(request.path, signal) if (request.action === 'reveal') await this.revealPath(path, signal) else if (request.application !== undefined) await this.openFileApplication(path, request.application, signal) else await this.openPath(path, signal) return { opened: true } } catch (error: unknown) { if (signal.aborted) throw new RemoteError('gateway/cancelled', 'path open was aborted', {}) if (error instanceof RemoteError) throw error throw new RemoteError( 'gateway/internal', 'path open failed', {}, { cause: error }, ) } } /** * Query current file handlers on the serving desktop without activating an Agent. * @param request - file path in Host filesystem syntax. * @param signal - caller lifetime, propagated to filesystem and desktop queries. * @returns OS application names, icons, and default selection; empty when desktop opening is unavailable. * @throws RemoteError when the path is invalid, the query is cancelled, or native discovery fails. */ @Remote('workspacePathApplications') async workspacePathApplications( request: { readonly path: string }, signal: AbortSignal, ): Promise { if (!this.canOpenPath()) return [] try { const path = await this.verifyDesktopPath(request.path, signal) return await this.fileApplications(path, signal) } catch (error: unknown) { if (signal.aborted) throw new RemoteError('gateway/cancelled', 'application query was aborted', {}) if (error instanceof RemoteError) throw error throw new RemoteError('gateway/internal', 'file application query failed', {}, { cause: error }) } } private async verifyDesktopPath(path: string, signal: AbortSignal): Promise { if (path.length === 0) throw new RemoteError('gateway/bad-request', 'A non-empty file path is required', {}) signal.throwIfAborted() const hostPath = resolve(path) const { fs } = this.ctx const mapped = fs.processPathFromHostPath(hostPath) if (mapped === undefined || fs.processPath(await fs.resolve(mapped, { signal })) !== hostPath) { throw new RemoteError('gateway/bad-request', 'Path has no verified Host path', {}) } signal.throwIfAborted() return hostPath } /** * Rename one Session after explicitly resuming it. * @param request - Session identity and proposed title. * @returns the accepted title and durable event sequence. */ @Remote('rename') rename(request: SessionRenameRequest): Promise { return this.commands.rename(request) } /** * Fork one cold-readable exact event prefix into a new Session. An omitted * boundary selects the latest completed-turn prefix; an open cut receives * synthetic fork closers. * @param request - source Session, optional exact inclusive event boundary, and migration preflight choice. * @returns the new Session identity. */ @Remote('fork') async fork(request: SessionForkRequest): Promise { if (request.allowMigration === false && this.ctx.sessions.get(request.sessionId) === undefined) { const stored = await this.ctx.get('sessionPersistence')?.stat(request.sessionId) if (stored?.formatStatus === 'migration-required' && this.ctx.sessions.get(request.sessionId) === undefined) { throw new RemoteError('session/migration-required', 'Open the source Session to complete migration before forking', { sessionId: request.sessionId, }) } } return this.commands.fork(request) } /** * Admit one prompt after explicitly resuming its Session. * @param request - Session identity, prompt content, source metadata, and delivery mode. * @param signal - caller cancellation before prompt admission begins. * @returns acknowledgement that the Agent accepted the prompt. */ @Remote('prompt') async prompt(request: SessionPromptRequest, signal: AbortSignal): Promise { signal.throwIfAborted() if (!hasPromptContent(request.content)) { throw new RemoteError('gateway/bad-request', 'prompt content must include non-whitespace text or an attachment', {}) } const clientTimeZone = request.clientTimeZone === undefined ? undefined : canonicalClientTimeZone(request.clientTimeZone) if (request.clientTimeZone !== undefined && clientTimeZone === undefined) { throw new RemoteError( 'session/invalid-time-zone', 'clientTimeZone must be UTC or a valid IANA Area/Location name', { value: request.clientTimeZone }, ) } // The caller is recorded only when it is not the owner, so owner messages keep their exact source. const principal = principalOf(this.ctx.invocation) const source: MessageSource = { kind: 'user', rpcId: request.requestId, ...(clientTimeZone === undefined ? {} : { clientTimeZone }), ...(principal === undefined ? {} : { principal }), } return this.commands.prompt(request, source) } /** * Read one image proven reachable from the addressed Session log. * @param request - Session and attachment identities used for authorization. * @returns the durable attachment reference and base64-encoded bytes. */ @Remote('attachment') attachment(request: SessionAttachmentRequest): Promise { return this.commands.attachment(request) } /** * Mutate one still-pending queue occurrence, resuming a cold Agent first. * @param request - Session, queue item, and requested mutation. * @returns acknowledgement that the queue mutation was applied. */ @Remote('updateQueue') updateQueue(request: SessionUpdateQueueRequest): Promise { return this.commands.updateQueue(request) } /** * Cancel one active Agent turn without dropping its pending inbox. * @param request - Session whose active Agent turn is cancelled. * @returns acknowledgement that cancellation was requested. */ @Remote('cancel') cancel(request: SessionCancelRequest): SessionCancelValue { return this.commands.cancel(request) } /** * Read one cold-safe, message-aligned Session history page. * @param request - durable address, backward cursor, and page budget. * @param signal - cancellation for persistence reads. * @returns one chronological page. */ @Remote({ name: 'page', resources: { session: addressedSession } }) page(request: SessionPageRequest, signal: AbortSignal): Promise { return this.history.page(request, signal) } /** * Follow one Session log from its opening or resume cursor. * @param request - durable address and last committed sequence already held by the caller. * @param signal - cancellation owned by the Remote stream carrier. * @returns a complete opening snapshot followed by gap-free durable event * frames and optional cursorless assistant-stream frames. */ @Remote({ mode: 'stream', resources: { session: addressedSession } }) follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable { return this.history.follow(request, signal) } /** * Read exact projections, returning cached hints only when migration is required. * @param request - Session whose current values are required. * @param signal - cancellation for the Session read. * @returns a sequenced baseline, cached hints when migration is deferred, or null when absent. */ @Remote('projections') async projections(request: SessionProjectionsRequest, signal: AbortSignal): Promise { const { sessionId } = request if (sessionId.length === 0) { throw new RemoteError('gateway/bad-request', 'sessionId must not be empty', {}) } try { signal.throwIfAborted() const live = this.liveProjections(sessionId) if (live !== undefined) return live const stored = await this.ctx.get('sessionPersistence')?.stat(sessionId, { signal }) signal.throwIfAborted() const attached = this.liveProjections(sessionId) if (attached !== undefined) return attached if (stored === undefined) return null if (stored.formatStatus === 'migration-required') { const cached = this.ctx.get('sessionProjectionCache')?.cachedSnapshot(stored.header) return { kind: 'migration-required', values: { ...cached?.values } as SessionProjectionValues } } using observation = await this.ctx.sessionQuery.observeSession(sessionId, { signal }) const projections = observation.projections if (projections === undefined) { throw new RemoteError('session/projections-unavailable', 'Session projections are unavailable', {}) } return { kind: 'sequenced', asOfSeq: projections.asOfSeq, values: projections.values as SessionProjectionValues } } catch (error: unknown) { if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') { return this.liveProjections(sessionId) ?? null } if (signal.aborted || (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) { throw new RemoteError('gateway/cancelled', 'Session projection read was cancelled', {}, { cause: error }) } if (error instanceof RemoteError) throw error throw new RemoteError('gateway/internal', 'Session projection read failed', {}, { cause: error }) } } private liveProjections( sessionId: SessionId, ): Extract | undefined { const session = this.ctx.sessions.get(sessionId) if (session === undefined) return undefined const snapshot = this.ctx.sessionProjections.snapshot(session) return { kind: 'sequenced', asOfSeq: snapshot.asOfSeq, values: snapshot.values as SessionProjectionValues } } /** * Stream a complete live-control baseline followed by replacement frames. * @param signal - cancellation owned by the Remote stream carrier. * @returns one complete baseline followed by live replacement frames. */ @Remote({ mode: 'stream' }) async *control(signal: AbortSignal): AsyncIterable { for await (const frame of this.controlState.control(signal)) { switch (frame.type) { case 'baseline': { const projections = Object.fromEntries(Object.entries(frame.value.projections) .filter(([sessionId]) => this.ctx.remoteAccess.allows({ kind: 'session', id: sessionId as SessionId }))) yield { type: 'baseline', value: { ...frame.value, projections } } break } case 'projection': if (this.ctx.remoteAccess.allows({ kind: 'session', id: frame.sessionId })) yield frame break /* v8 ignore next -- closed control-frame union. */ default: assertNever(frame) } } } } export { buildModelCatalog } export default SessionController