diff --git a/packages/api/session-controller/package.json b/packages/api/session-controller/package.json index a60f4b3e8d..a6d764bf4a 100644 --- a/packages/api/session-controller/package.json +++ b/packages/api/session-controller/package.json @@ -84,6 +84,7 @@ "@deepseek-ai/dsh-api-gateway": "workspace:^", "@deepseek-ai/dsh-attachment": "workspace:^", "@deepseek-ai/dsh-brand": "workspace:^", + "@deepseek-ai/dsh-client-connection": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", "@deepseek-ai/dsh-jobs": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", @@ -115,6 +116,8 @@ "@deepseek-ai/dsh-api-gateway": "workspace:^", "@deepseek-ai/dsh-attachment": "workspace:^", "@deepseek-ai/dsh-brand": "workspace:^", + "@deepseek-ai/dsh-client-connection": "workspace:^", + "@deepseek-ai/dsh-client-store": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", "@deepseek-ai/dsh-jobs": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", @@ -130,6 +133,7 @@ "@deepseek-ai/dsh-tools": "workspace:^", "@deepseek-ai/dsh-typert-protocol": "workspace:^", "@deepseek-ai/dsh-typert-registry": "workspace:^", + "@deepseek-ai/dsh-util-crypto": "workspace:^", "@deepseek-ai/dsh-workspace": "workspace:^" } } diff --git a/packages/api/session-controller/src/client/contract/events.ts b/packages/api/session-controller/src/client/contract/events.ts new file mode 100644 index 0000000000..f71c4be0b0 --- /dev/null +++ b/packages/api/session-controller/src/client/contract/events.ts @@ -0,0 +1,82 @@ +/** Observable contiguous Session event window consumed by domain assemblers. */ +import { notifySubscribers, type ObservableSnapshot } from '@deepseek-ai/dsh-client-store' +import type { SessionEventEntry } from '../../types.ts' + +/** Exact delta that produced the latest event-window revision. */ +export type SessionEventChange = + | { readonly kind: 'replace'; readonly entries: readonly SessionEventEntry[] } + | { readonly kind: 'prepend'; readonly entries: readonly SessionEventEntry[] } + | { readonly kind: 'append'; readonly entries: readonly SessionEventEntry[] } + +/** Current contiguous event window and its latest synchronous delta. */ +export interface SessionEventWindow { + readonly entries: readonly SessionEventEntry[] + readonly hasMore: boolean + readonly revision: number + readonly change: SessionEventChange +} + +/** Conversation-facing event source exposed by one Session binding. */ +export type SessionEventSource = ObservableSnapshot + +/** Session-owned event feed; every accepted window mutation publishes synchronously. */ +export class MutableSessionEventSource implements SessionEventSource { + private readonly listeners = new Set<() => void>() + private snapshot: SessionEventWindow = { + entries: [], + hasMore: false, + revision: 0, + change: { kind: 'replace', entries: [] }, + } + + /** @returns the cached event-window snapshot. */ + getSnapshot(): SessionEventWindow { return this.snapshot } + + /** + * Subscribe to synchronous window publication. + * @param listener - invalidation callback. + * @returns unsubscribe function. + */ + subscribe(listener: () => void): () => void { + this.listeners.add(listener) + return () => { this.listeners.delete(listener) } + } + + /** + * Replace the complete contiguous window. + * @param entries - complete window. + * @param hasMore - whether older history remains. + */ + replace(entries: readonly SessionEventEntry[], hasMore: boolean): void { + this.publish(entries, hasMore, { kind: 'replace', entries }) + } + + /** + * Prepend one older contiguous page. + * @param entries - newly loaded older entries. + * @param hasMore - whether still older history remains. + */ + prepend(entries: readonly SessionEventEntry[], hasMore: boolean): void { + this.publish([...entries, ...this.snapshot.entries], hasMore, { kind: 'prepend', entries }) + } + + /** + * Append one contiguous live entry. + * @param entry - live tail entry. + */ + append(entry: SessionEventEntry): void { + this.publish([...this.snapshot.entries, entry], this.snapshot.hasMore, { + kind: 'append', + entries: [entry], + }) + } + + private publish( + entries: readonly SessionEventEntry[], + hasMore: boolean, + change: SessionEventChange, + ): void { + this.snapshot = { entries, hasMore, revision: this.snapshot.revision + 1, change } + notifySubscribers(this.listeners, '[session-controller] event feed') + } +} diff --git a/packages/api/session-controller/src/client/contract/result.ts b/packages/api/session-controller/src/client/contract/result.ts new file mode 100644 index 0000000000..6577b63823 --- /dev/null +++ b/packages/api/session-controller/src/client/contract/result.ts @@ -0,0 +1,28 @@ +/** Client operation results spanning Session Remote calls and the legacy subagent carrier. */ + +import type { RpcError } from '@deepseek-ai/dsh-client-connection/client' +import type { SessionError } from '../../types.ts' + +/** Failure surfaced by the Client Session object layer. */ +export type ClientFailure = RpcError | SessionError + +/** Success or failure returned by a Client Session operation. */ +export type ClientResult = + | { readonly ok: true; readonly value: T } + | { readonly ok: false; readonly error: ClientFailure } + +/** + * Fold a rejected carrier operation into the Client Session failure vocabulary. + * @param error - rejection from a legacy subagent or local carrier call. + * @returns the failure branch of a Client Session result. + */ +export function transportResult(error: unknown): ClientResult { + return { + ok: false, + error: { + code: 'internal', + message: error instanceof Error ? error.message : String(error), + details: {}, + }, + } +} diff --git a/packages/client/runtime/src/client/contract/session.ts b/packages/api/session-controller/src/client/contract/session.ts similarity index 84% rename from packages/client/runtime/src/client/contract/session.ts rename to packages/api/session-controller/src/client/contract/session.ts index aff25a1478..6ac4182bb9 100644 --- a/packages/client/runtime/src/client/contract/session.ts +++ b/packages/api/session-controller/src/client/contract/session.ts @@ -1,19 +1,20 @@ /** * The outward session face. Feature packages never see the concrete Session - * class: components read conversation state through `useSession` (the + * class: components read lifecycle state through `useSession` (the * ObservableSnapshot half), and orchestration code calls the behavior verbs * below — nothing else. Widening this interface is the explicit act of * widening what features may do to a session (and what every test fixture - * must stub); runtime-internal entry points (history staging, wire-frame + * must stub); implementation-internal entry points (history staging, wire-frame * dispatch) stay on the class, invisible out here. */ import type { AttachmentIdType, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' -import type { - ClientResult, MessageId, PromptContentPart, QueueAction, SessionId, -} from '@deepseek-ai/dsh-api-remotes/client' +import type { MessageId } from '@deepseek-ai/dsh-llm/brand' +import type { SessionId } from '@deepseek-ai/dsh-session/types' import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol' -import type { ConversationSnapshot } from '../sessions/conversation.ts' -import type { ObservableSnapshot } from './store.ts' +import type { ObservableSnapshot } from '@deepseek-ai/dsh-client-store' +import type { PromptContentPart, QueueAction } from '../../types.ts' +import type { ClientResult } from './result.ts' +import type { SessionSnapshot } from './snapshot.ts' /** Key-addressed projection read face (the useProjection resolution path; see ProjectionValueStore). */ export interface ProjectionsFace { @@ -86,8 +87,8 @@ export interface ISession { } /** - * The full outward face: behavior verbs plus the conversation read side + * The full outward face: behavior verbs plus the Session lifecycle read side * (the `useSession` hook source). This is the type carried by * `SessionBinding.session` and the provide channel. */ -export type SessionFace = ISession & ObservableSnapshot +export type SessionFace = ISession & ObservableSnapshot diff --git a/packages/client/runtime/src/client/contract/sessions.ts b/packages/api/session-controller/src/client/contract/sessions.ts similarity index 78% rename from packages/client/runtime/src/client/contract/sessions.ts rename to packages/api/session-controller/src/client/contract/sessions.ts index ca53adf037..4e64574681 100644 --- a/packages/client/runtime/src/client/contract/sessions.ts +++ b/packages/api/session-controller/src/client/contract/sessions.ts @@ -1,39 +1,42 @@ /** * The outward sessions-service face — what `ctx.sessions` exposes to feature - * packages and the renderer host, and therefore exactly what the test - * runtime's sessions double must implement. Transport entry points and - * runtime internals stay on - * the concrete class; cross-domain consumers keep the narrower - * [SessionsPort](./sessions-port.ts). Widening this interface is the + * packages. Transport entry points and implementation internals stay on + * the concrete class. Widening this interface is the * explicit act of widening what features may do to the sessions domain. */ import type { Context } from '@deepseek-ai/cordis' -import type { - ClientResult, SessionId, SubagentAddress, -} from '@deepseek-ai/dsh-api-remotes/client' -import type { HostObservable, SessionMaybeProvideInfo } from '@deepseek-ai/dsh-client-ui-slots' -import type { AgentContext } from '../agents/scope.ts' +import type { SubagentAddress } from '@deepseek-ai/dsh-client-connection/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types' +import type { AgentContext } from '../scope.ts' import type { SessionSearchResultItem } from '../sessions/manager.ts' -import type { - SessionBinding, SessionListState, SessionProvideDescriptor, -} from '../sessions/service.ts' +import type { SessionBinding, SessionListState } from '../sessions/service.ts' +import type { ClientResult } from './result.ts' import type { SessionFace } from './session.ts' -import type { ObservableSnapshot } from './store.ts' +import type { ObservableSnapshot } from '@deepseek-ai/dsh-client-store' -export type { AgentContext } from '../agents/scope.ts' +export type { AgentContext } from '../scope.ts' /** The sessions-service face injected as `ctx.sessions`. */ export interface ISessions { /** The useSessions standard feed (list rows + current selection; read face — writes stay inside the domain). */ readonly list: ObservableSnapshot - /** Atomic current-session provide projection (the renderer host's `sessions.provideInfo` feed). */ - readonly currentProvideInfo: HostObservable /** * The `session.search` result bound the wire schema fixes, exposed to * presentation as injected data. Not per-connection state: every transport * (fixture included) reports the same number. */ readonly searchResultLimit: number + /** + * Create or adopt a Session on the Host. + * @param opts - target workspace, directory, and optional preallocated identity. + * @returns the Session identity after its local binding is addressable. + */ + create(opts?: { + workspaceId?: WorkspaceId + cwd?: string + sessionId?: SessionId + }): Promise /** * Select a session as current. * @param id - session id (must exist in the list; unknown ids fail loud). @@ -73,6 +76,8 @@ export interface ISessions { noteAgentPreset(sessionId: SessionId, agentPreset: string): void /** Clear the current selection into the no-session view state. */ clear(): void + /** @returns completion of the current or newly started Session-list refresh. */ + refresh(): Promise /** * Search the Host's visible message-content index. Results stay * request-local; the list snapshot remains the metadata authority. @@ -95,13 +100,6 @@ export interface ISessions { * @throws when the fork fails, or when a requested child-title rename fails after creation. */ fork(opts: { sessionId: SessionId; atSeq?: number; increaseTitle?: boolean }): Promise - /** - * Register a per-session standard-props provider (hooks become `use` - * selector hooks on the render side; props spread verbatim). - * @param descriptor - static member roster plus per-session resolver. - * @returns disposer removing the provider. - */ - provide(descriptor: SessionProvideDescriptor): () => void /** * Resolve an Agent-scoped context view (use-and-discard). * @param id - session id. diff --git a/packages/api/session-controller/src/client/contract/snapshot.ts b/packages/api/session-controller/src/client/contract/snapshot.ts new file mode 100644 index 0000000000..7304c21bb8 --- /dev/null +++ b/packages/api/session-controller/src/client/contract/snapshot.ts @@ -0,0 +1,45 @@ +/** Session-owned observable state excluding Conversation target data. */ +import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' +import type { MessageId } from '@deepseek-ai/dsh-llm/brand' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import type { SubagentAddress } from '@deepseek-ai/dsh-client-connection/client' +import type { ClientFailure } from './result.ts' + +/** One transient inbox occurrence from the authoritative queue snapshot. */ +export interface QueuedMessage { + readonly id: MessageId + readonly messageId: MessageId + readonly placement: 'queued' | 'steering' | 'context' + readonly content: readonly ContentBlock[] + readonly preview: string + readonly text: string | null +} + +/** History-open lifecycle of a Session event window. */ +export type OpenState = 'cold' | 'loading' | 'open' | 'error' + +/** Send/stop failure surfaced by Session consumers. */ +export interface PromptError { + readonly op: 'send' | 'stop' + readonly error: ClientFailure +} + +/** Immutable Session lifecycle and control snapshot. */ +export interface SessionSnapshot { + readonly sessionId: SessionId + readonly queue: readonly QueuedMessage[] + readonly running: boolean + readonly subagent: { readonly address: SubagentAddress; readonly parentAvailable: boolean } | null + readonly removed: boolean + readonly openState: OpenState + readonly openError: ClientFailure | null + readonly hasMore: boolean + readonly loadingOlder: boolean + readonly promptError: PromptError | null + readonly blank: boolean + readonly lastAgentError: string | null + /** A prompt call has begun on this Client Session object. */ + readonly promptAttempted: boolean + /** The first accepted prompt has not reached a durable `turn/start` event. */ + readonly awaitingFirstTurn: boolean +} diff --git a/packages/api/session-controller/src/client/index.ts b/packages/api/session-controller/src/client/index.ts index 7c165ba043..d5dae15109 100644 --- a/packages/api/session-controller/src/client/index.ts +++ b/packages/api/session-controller/src/client/index.ts @@ -1,184 +1,125 @@ -/** Session-specific adapters for Gateway-owned Remote stream lifecycles. */ +/** Client Session object layer, Agent scopes, and Remote lifecycle wiring. */ -import type {} from '@deepseek-ai/dsh-api-session-controller/remote' -import type { RemoteFailure } from '@deepseek-ai/dsh-typert-protocol' -import { - RemoteJournalStream, - RemoteSnapshotStream, - RemoteStreamCarrierError, - RemoteStreamError, - type ClientRemote, - type RemoteJournalChange, - type RemoteJournalFrame, -} from '@deepseek-ai/dsh-api-gateway/client' -import type { - SessionAddress, - SessionControlFrame, - SessionEventEntry, - SessionPage, - SessionPageRequest, -} from '../types.ts' +import type { Context } from '@deepseek-ai/cordis' +import type {} from '@deepseek-ai/dsh-agent/types' +import type { ConnectionHandle } from '@deepseek-ai/dsh-client-connection/client' +import { createSessionControlStream } from './transport.ts' +import { ClientSessions } from './sessions/service.ts' +import type { ISessions } from './contract/sessions.ts' +import type { SessionRemotes } from './sessions/remotes.ts' +import type {} from '../remote-events.ts' export { + createSessionControlStream, + SessionEventStream, SESSION_SEARCH_RESULT_LIMIT, SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS, -} from '../types.ts' + sessionStreamFailure, +} from './transport.ts' +export type { + ClientSessionPageRequest, + SessionControlStream, + SessionControlStreamOptions, + SessionEventStreamOptions, + SessionJournalChange, + SessionRemote, +} from './transport.ts' +export { createScope, scopeOf } from './scope.ts' +export type { AgentContext, AgentScopeHandle } from './scope.ts' +export { SessionCreateError, SessionForkError, workspaceTitleOf } from './sessions/service.ts' +export type { SessionBinding, SessionListState, SessionSummary } from './sessions/service.ts' +export type { + SessionListPhase, + SessionListSnapshot, + SessionSearchResultItem, + SubagentCatalogSnapshot, +} from './sessions/manager.ts' +export type { Session } from './sessions/session.ts' +export type { + ProjectionsBaseline, + ProjectionValueStore, + SessionProjectionMap, + UseProjection, +} from './sessions/projection-store.ts' +export type { ISession, ProjectionsFace, SessionFace } from './contract/session.ts' +export type { ISessions } from './contract/sessions.ts' +export { MutableSessionEventSource } from './contract/events.ts' +export type { SessionEventChange, SessionEventSource, SessionEventWindow } from './contract/events.ts' +export type { + OpenState, + PromptError, + QueuedMessage, + SessionSnapshot, +} from './contract/snapshot.ts' +export type { ClientFailure, ClientResult } from './contract/result.ts' +export { indexSubagentDescendants } from './sessions/subagent-lineage.ts' +export type { SubagentDescendantSummary } from './sessions/subagent-lineage.ts' -/** Session Controller's Client row exports library values and installs no Cordis service. */ -export function apply(): void {} +declare module '@deepseek-ai/cordis' { + interface Events { + /** + * A Host connection generation completed its readiness handshake. + * @mode emit + */ + 'connection/reset'(): void + } -/** Pagination fields bound to an already-addressed Session journal. */ -export type ClientSessionPageRequest = Omit - -/** Complete generated `ctx.remote.session` namespace. */ -export type SessionRemote = ClientRemote['session'] - -/** One complete event-window publication from the Session journal stream. */ -export type SessionEventChange = RemoteJournalChange - -type SessionControlBaselineFrame = Extract -type SessionControlDeltaFrame = Exclude - -/** Gateway-owned control snapshot stream configured for Session frames. */ -export type SessionControlStream = RemoteSnapshotStream< - SessionControlBaselineFrame, - SessionControlDeltaFrame -> - -type SessionStreamRemote = Pick - -/** Domain sinks used by the Host-wide Session control stream. */ -export interface SessionControlStreamOptions { - /** Apply a complete baseline or one later update. */ - readonly accept: (frame: SessionControlFrame) => void - /** Observe a retryable carrier loss before reconnection. */ - readonly carrierFailed?: (error: RemoteStreamCarrierError) => void - /** Publish a terminal business or protocol failure. */ - readonly failed: (error: unknown) => void + interface Context { + /** Client Session object layer and Agent scope owner. */ + sessions: import('./contract/sessions.ts').ISessions + } } -/** Domain sinks used by one addressed Session event journal. */ -export interface SessionEventStreamOptions { - /** Apply one complete event-window change. */ - readonly publish: (change: SessionEventChange) => void - /** Observe a retryable carrier loss before reconnection. */ - readonly carrierFailed?: (error: RemoteStreamCarrierError) => void - /** Publish a terminal stream, page, or protocol failure after opening. */ - readonly failed: (error: unknown) => void +/** Required wire, Remote, and Context projection services. */ +export const inject = [ + 'connection', + 'typert', + 'remote', + 'remote.commands', + 'remote.session', +] + +/** + * Resolve the Client Session service from any Client Cordis context. + * @param ctx - Client root or Agent-scoped context. + * @returns the Client Session object layer. + */ +export function resolveClientSessions(ctx: Context): ISessions { + const sessions = ctx.get('sessions') + if (sessions === undefined) throw new Error('session-controller: Client sessions service unavailable') + return sessions } /** - * Create the Host-wide Session control snapshot stream. - * @param remote - generated Session namespace and Gateway stream factory. - * @param options - Session state destinations. - * @returns an unstarted stream owned by the Client Session runtime. + * Install Client Session state and its reconnecting control stream. + * @param ctx - Client Cordis context. */ -export function createSessionControlStream( - remote: SessionStreamRemote, - options: SessionControlStreamOptions, -): SessionControlStream { - const stream = remote.$stream({ - name: 'session control stream', - open: signal => remote.session.control(signal), - ended: accepted => accepted - ? new RemoteStreamCarrierError('session control stream ended without a terminal result') - : new Error('session control stream ended before its opening snapshot'), - ...(options.carrierFailed === undefined ? {} : { carrierFailed: options.carrierFailed }), +export function apply(ctx: Context): void { + const connection = ctx.get('connection') as ConnectionHandle + const remotes = ctx.remote as unknown as SessionRemotes + const sessions = new ClientSessions(ctx, connection.api, remotes) + ctx.remote.$on('api-session/added', (summary) => { sessions.handleSessionAdded(summary) }) + ctx.remote.$on('api-session/removed', (sessionId) => { sessions.handleSessionRemoved(sessionId) }) + ctx.remote.$on('api-session/status', (sessionId, running) => { + sessions.handleSessionStatus(sessionId, running) }) - return new RemoteSnapshotStream(stream, { - name: 'session control stream', - isSnapshot: (frame): frame is SessionControlBaselineFrame => frame.type === 'baseline', - replace: options.accept, - update: options.accept, - failed: options.failed, + ctx.remote.$on('api-session/activity', (sessionId, updatedAt) => { + sessions.handleSessionActivity(sessionId, updatedAt) + }) + ctx.remote.$on('api-session/error', (sessionId, message) => { + sessions.handleSessionError(sessionId, message) }) -} -/** Gateway-owned event journal bound to one ordinary or direct-subagent Session address. */ -export class SessionEventStream extends RemoteJournalStream< - SessionPage, - SessionEventEntry, - number, - ClientSessionPageRequest -> { - /** - * @param remote - generated Session namespace and Gateway stream factory. - * @param address - durable ordinary-Session or direct-subagent address. - * @param options - Session event-window destinations. - */ - constructor( - private readonly remote: SessionStreamRemote, - private readonly address: SessionAddress, - options: SessionEventStreamOptions, - ) { - super(remote, { - name: 'session event stream', - emptyCursor: -1, - entries: page => page.events, - hasMore: page => page.hasMore, - cursor: entry => entry.event.seq, - compare: (left, right) => left - right, - follows: (left, right) => right === left + 1, - publish: options.publish, - ...(options.carrierFailed === undefined - ? {} - : { carrierFailed: options.carrierFailed }), - failed: options.failed, - }) - } - - /** @inheritdoc */ - protected override async * follow( - afterSeq: number | undefined, - signal: AbortSignal, - ): AsyncIterable> { - const request = afterSeq === undefined - ? { address: this.address } - : { address: this.address, afterSeq } - for await (const frame of this.remote.session.follow(request, signal)) { - if (frame.type === 'opened') { - yield frame - continue - } - const { type: _type, ...entry } = frame - yield { type: 'entry', entry } - } - } - - /** @inheritdoc */ - protected override async readPage( - request: ClientSessionPageRequest, - throughSeq: number, - signal: AbortSignal, - ): Promise { - const result = await this.remote.session.page( - { address: this.address, throughSeq, ...request }, - signal, - ) - if (!result.ok) { - throw new RemoteStreamError( - result.error.code, - result.error.message, - result.error.details, - ) - } - return result.value - } - - /** @inheritdoc */ - protected override repairRequest( - request: ClientSessionPageRequest, - ): ClientSessionPageRequest { - return request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages } - } -} - -/** - * Recover a Host Session failure from a Remote stream terminal error. - * @param error - value thrown while opening or consuming a Session stream. - * @returns the Host failure, or `undefined` for carrier and local failures. - */ -export function sessionStreamFailure(error: unknown): RemoteFailure | undefined { - if (!(error instanceof RemoteStreamError)) return undefined - return { code: error.code, message: error.message, details: error.details } + const control = createSessionControlStream(remotes, { + accept: (frame) => { sessions.handleControlFrame(frame) }, + failed: (error) => { console.error('[session-controller] control stream failed:', error) }, + }) + control.start() + ctx.on('connection/reset', () => { sessions.handleConnected() }) + if (connection.hostDescription.getSnapshot() !== undefined) sessions.handleConnected() + ctx.typert.contexts.registerClient('agent', { + identity: candidate => sessions.scopeOf(candidate), + resolve: sessionId => sessions.scope(sessionId), + }) + ctx.effect(() => async () => { await control.dispose() }, 'session-controller.client.control') } diff --git a/packages/client/runtime/src/client/ordered-baseline.ts b/packages/api/session-controller/src/client/ordered-baseline.ts similarity index 100% rename from packages/client/runtime/src/client/ordered-baseline.ts rename to packages/api/session-controller/src/client/ordered-baseline.ts diff --git a/packages/client/runtime/src/client/agents/scope.ts b/packages/api/session-controller/src/client/scope.ts similarity index 98% rename from packages/client/runtime/src/client/agents/scope.ts rename to packages/api/session-controller/src/client/scope.ts index 8e88ef6ebc..5e1dca62a3 100644 --- a/packages/client/runtime/src/client/agents/scope.ts +++ b/packages/api/session-controller/src/client/scope.ts @@ -17,8 +17,8 @@ */ import { Context as CordisContext } from '@deepseek-ai/cordis' import type { Context, Fiber } from '@deepseek-ai/cordis' -import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client' import type { ClientRemote } from '@deepseek-ai/dsh-api-gateway/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' import type { TypertRemoteScopeApi } from '@deepseek-ai/dsh-typert-protocol' /** Client Cordis Context carrying one Agent identity and its scoped Remote namespaces. */ diff --git a/packages/client/runtime/src/client/sessions/lineage.ts b/packages/api/session-controller/src/client/sessions/lineage.ts similarity index 94% rename from packages/client/runtime/src/client/sessions/lineage.ts rename to packages/api/session-controller/src/client/sessions/lineage.ts index 5b703baff8..3f6e21dbc9 100644 --- a/packages/client/runtime/src/client/sessions/lineage.ts +++ b/packages/api/session-controller/src/client/sessions/lineage.ts @@ -2,8 +2,9 @@ // The input order is authoritative; lineage only makes each child adjacent to its parent. // Orphaned lineage degrades to root level; cycles fail soft and emit as roots. -import type { SessionId, SessionSummary } from '@deepseek-ai/dsh-api-remotes/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types' +import type { SessionSummary } from '../../types.ts' /** Host list summary enriched with the latest Session Controller title projection. */ export interface TitledSessionSummary extends SessionSummary { @@ -65,7 +66,7 @@ export function flattenLineage( const visited = new Set() const walk = (s: TitledSessionSummary, depth: number): void => { if (visited.has(s.sessionId)) { - console.warn(`[web-runtime] lineage cycle at ${s.sessionId}; emitting as root`) + console.warn(`[session-controller] lineage cycle at ${s.sessionId}; emitting as root`) return } visited.add(s.sessionId) diff --git a/packages/client/runtime/src/client/sessions/manager.ts b/packages/api/session-controller/src/client/sessions/manager.ts similarity index 95% rename from packages/client/runtime/src/client/sessions/manager.ts rename to packages/api/session-controller/src/client/sessions/manager.ts index ad111dd391..ddfae1d9e2 100644 --- a/packages/client/runtime/src/client/sessions/manager.ts +++ b/packages/api/session-controller/src/client/sessions/manager.ts @@ -1,22 +1,23 @@ // SessionManager: the instance cluster Map (lazy-built, resident) + the frame -// dispatch entry + list state, constructed and held by SessionRuntime (one per client runtime). +// dispatch entry + list state, constructed and held by ClientSessions (one per browser client). // List data never enters zustand; React connects via subscribe/getListSnapshot. import type { - ClientFailure, ClientResult, IApiClient, SessionId, - SessionSummary, SubagentAddress, SubagentCatalog, JobView, WorkspaceId, -} from '@deepseek-ai/dsh-api-remotes/client' + IApiClient, SubagentAddress, SubagentCatalog, +} from '@deepseek-ai/dsh-client-connection/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types' import type { SessionControlBaseline, SessionControlFrame, SessionQueuedItem, SessionError, -} from '@deepseek-ai/dsh-api-session-controller/types' -// Value import from the inline-safe wire layer (not the connection plugin): -// plugin-to-plugin value imports are a bundle purity error. -import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api' + SessionSummary, + SessionJob as JobView, +} from '../../types.ts' import { mergeOrderedBaseline } from '../ordered-baseline.ts' -import type { ConversationRuntime } from './conversation-assembler.ts' +import type { ClientFailure, ClientResult } from '../contract/result.ts' +import { transportResult } from '../contract/result.ts' import type { SessionListEntry, TitledSessionSummary } from './lineage.ts' import { flattenLineage } from './lineage.ts' // Type-only merge edge: the title domain's client-namespace outlet declares @@ -84,6 +85,8 @@ type SessionListMutation = /** Instance cluster + frame entry + the session list. */ export class SessionManager { private readonly sessions = new Map() + /** In-flight Session disposals remain here after instances leave `sessions`, so manager disposal can await quiescence. */ + private readonly sessionDisposals = new Set>() /** Latest transient queues, retained independently of Session object materialization. */ private readonly queues = new Map() /** @@ -142,7 +145,6 @@ export class SessionManager { private readonly remote: SessionRemotes, restoredSelection?: SessionId, restoredAddress?: SubagentAddress, - private readonly conversation?: ConversationRuntime, ) { this.selected = restoredSelection if (restoredAddress !== undefined) this.addresses.set(restoredAddress.childSessionId, restoredAddress) @@ -232,11 +234,41 @@ export class SessionManager { * truth — a later get() lazily rebuilds and open() backfills history. * @param sessionId - the session to drop. */ - drop(sessionId: SessionId): Promise { + async drop(sessionId: SessionId): Promise { const session = this.sessions.get(sessionId) - if (session === undefined) return Promise.resolve() this.sessions.delete(sessionId) - return session.dispose() + if (session !== undefined) await this.startSessionDisposal(session) + } + + /** + * Stop owned timers and every remaining Session instance. + * @returns when every Session Remote iterator has completed teardown. + */ + async dispose(): Promise { + for (const timer of this.catalogDebounce.values()) clearTimeout(timer) + this.catalogDebounce.clear() + this.catalogStale.clear() + this.openCatalogs.clear() + const sessions = [...this.sessions.values()] + this.sessions.clear() + for (const session of sessions) void this.startSessionDisposal(session) + await this.drainSessionDisposals() + } + + private startSessionDisposal(session: Session): Promise { + const disposal = session.dispose() + this.sessionDisposals.add(disposal) + void disposal.then( + () => { this.sessionDisposals.delete(disposal) }, + () => { this.sessionDisposals.delete(disposal) }, + ) + return disposal + } + + private async drainSessionDisposals(): Promise { + while (this.sessionDisposals.size > 0) { + await Promise.allSettled([...this.sessionDisposals]) + } } /** @@ -289,15 +321,9 @@ export class SessionManager { this.recordMutation({ kind: 'engaged', sessionId: engaged.sessionId }) }, projections: this.projectionStore(sessionId), - ...this.conversation === undefined ? {} : { conversation: this.conversation }, }) } - /** Rebuild every resident Session after one coalesced registry transaction. */ - rebuildConversationRegistry(): void { - for (const session of this.sessions.values()) session.rebuildConversationRegistry() - } - /** Resident per-session projection store (create-on-demand; outlives instantiation). */ private projectionStore(sessionId: SessionId): ProjectionValueStore { let store = this.projectionStores.get(sessionId) @@ -357,7 +383,7 @@ export class SessionManager { }) } } catch (error: unknown) { - const folded = transportError(error) + const folded = transportResult(error) this.catalogs.set(parentSessionId, { entries: this.withCatalogMutations( previous?.entries ?? [], expandableRows, activityRows, @@ -467,8 +493,8 @@ export class SessionManager { } } catch (error) { this.listState = 'error' - const folded = transportError(error) - /* v8 ignore next -- the `? null` arm is unreachable: transportError always returns ok:false. */ + const folded = transportResult(error) + /* v8 ignore next -- the `? null` arm is unreachable: transportResult always returns ok:false. */ this.listError = folded.ok ? null : folded.error } finally { this.listMutations = null @@ -501,7 +527,7 @@ export class SessionManager { }, } } catch (error: unknown) { - return transportError(error) + return transportResult(error) } } @@ -547,7 +573,7 @@ export class SessionManager { } return result } catch (error) { - return transportError(error) + return transportResult(error) } } @@ -581,7 +607,7 @@ export class SessionManager { } return result } catch (error) { - return transportError(error) + return transportResult(error) } } diff --git a/packages/client/runtime/src/client/sessions/notifier.ts b/packages/api/session-controller/src/client/sessions/notifier.ts similarity index 95% rename from packages/client/runtime/src/client/sessions/notifier.ts rename to packages/api/session-controller/src/client/sessions/notifier.ts index cc15beebd7..660c8645b4 100644 --- a/packages/client/runtime/src/client/sessions/notifier.ts +++ b/packages/api/session-controller/src/client/sessions/notifier.ts @@ -1,3 +1,5 @@ +import { notifySubscribers } from '@deepseek-ai/dsh-client-store' + /** * Batches structural updates in microtasks and stream updates by animation * frame. Reads may rebuild a dirty snapshot without consuming the pending @@ -90,6 +92,6 @@ export class Notifier { this.dirty = false this.rebuild() } - for (const listener of this.listeners) listener() + notifySubscribers(this.listeners, '[session-controller]') } } diff --git a/packages/client/runtime/src/client/sessions/projection-store.ts b/packages/api/session-controller/src/client/sessions/projection-store.ts similarity index 99% rename from packages/client/runtime/src/client/sessions/projection-store.ts rename to packages/api/session-controller/src/client/sessions/projection-store.ts index 8996d73c78..4ffa2368bb 100644 --- a/packages/client/runtime/src/client/sessions/projection-store.ts +++ b/packages/api/session-controller/src/client/sessions/projection-store.ts @@ -9,7 +9,7 @@ * bare observable faces feed `useProjection` (ui-renderer binds them). */ import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types' -import type { ObservableSnapshot } from '../contract/store.ts' +import type { ObservableSnapshot } from '@deepseek-ai/dsh-client-store' import { Notifier } from './notifier.ts' // The single projection type table, typed end to end (host unit, wire block, diff --git a/packages/client/runtime/src/client/sessions/queue-mirror.ts b/packages/api/session-controller/src/client/sessions/queue-mirror.ts similarity index 93% rename from packages/client/runtime/src/client/sessions/queue-mirror.ts rename to packages/api/session-controller/src/client/sessions/queue-mirror.ts index b364f5b104..209349af2c 100644 --- a/packages/client/runtime/src/client/sessions/queue-mirror.ts +++ b/packages/api/session-controller/src/client/sessions/queue-mirror.ts @@ -1,7 +1,7 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' -import type { SessionQueuedItem } from '@deepseek-ai/dsh-api-session-controller/types' +import type { SessionQueuedItem } from '../../types.ts' import type { SessionEvent } from '@deepseek-ai/dsh-session/types' -import type { QueuedMessage } from './conversation.ts' +import type { QueuedMessage } from '../contract/snapshot.ts' const QUEUE_PREVIEW_CHARS = 200 diff --git a/packages/api/session-controller/src/client/sessions/remotes.ts b/packages/api/session-controller/src/client/sessions/remotes.ts new file mode 100644 index 0000000000..1449cc2c83 --- /dev/null +++ b/packages/api/session-controller/src/client/sessions/remotes.ts @@ -0,0 +1,29 @@ +/** + * Remote namespaces the Session cluster calls. One parameter for one concept: + * the generated surface a Session and its manager reach the Host through. + * + * @module @deepseek-ai/dsh-api-session-controller/client/sessions/remotes + */ + +import type { EncodedImageAttachment } from '@deepseek-ai/dsh-attachment/types' +import type { ClientRemote } from '@deepseek-ai/dsh-api-gateway/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol' +import type { SessionRemote } from '../transport.ts' + +/** Narrow Commands namespace consumed by a Client Session. */ +export interface SessionCommandsRemote { + execute( + agentId: SessionId, + line: string, + images: readonly EncodedImageAttachment[], + signal?: AbortSignal, + ): Promise> +} + +/** Generated Remote namespaces consumed by the Client Session object layer. */ +export interface SessionRemotes { + readonly $stream: ClientRemote['$stream'] + readonly commands: SessionCommandsRemote + readonly session: SessionRemote +} diff --git a/packages/client/runtime/src/client/sessions/service.ts b/packages/api/session-controller/src/client/sessions/service.ts similarity index 77% rename from packages/client/runtime/src/client/sessions/service.ts rename to packages/api/session-controller/src/client/sessions/service.ts index 7f1dc908d2..d2594f600f 100644 --- a/packages/client/runtime/src/client/sessions/service.ts +++ b/packages/api/session-controller/src/client/sessions/service.ts @@ -1,5 +1,5 @@ /** - * SessionRuntime: root sessions service — list snapshot store (manager + * ClientSessions: root sessions service — list snapshot store (manager * projection; carries `current`, the persisted selection every * session-scoped surface keys off), Agent scope tree (mintScope pattern: no-op plugin * Fiber + ctx.extend scope tag; one scope per session, agent id === session @@ -16,23 +16,24 @@ */ import type { Context, Fiber } from '@deepseek-ai/cordis' import type { - ClientFailure, ClientResult, IApiClient, SessionId, SubagentAddress, JobView, WorkspaceId, -} from '@deepseek-ai/dsh-api-remotes/client' -import { SESSION_SEARCH_RESULT_LIMIT } from '@deepseek-ai/dsh-api-session-controller/client' -import type { - HostObservable, SessionMaybeProvideInfo, SessionProvideInfo, -} from '@deepseek-ai/dsh-client-ui-slots' + IApiClient, SubagentAddress, +} from '@deepseek-ai/dsh-client-connection/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types' +import { SESSION_SEARCH_RESULT_LIMIT } from '../../types.ts' +import type { SessionJob as JobView } from '../../types.ts' import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types' -import type { SnapshotStore } from '../contract/store.ts' -import { createSnapshotStore } from '../contract/store.ts' +import { + createSnapshotStore, type SnapshotStore, +} from '@deepseek-ai/dsh-client-store' +import type { ClientFailure, ClientResult } from '../contract/result.ts' +import type { SessionEventSource } from '../contract/events.ts' import type { SessionFace } from '../contract/session.ts' import type { AgentContext, ISessions } from '../contract/sessions.ts' -import { createScope, scopeOf as scopeTagOf } from '../agents/scope.ts' -import type { ConversationRuntime } from './conversation-assembler.ts' +import { createScope, scopeOf as scopeTagOf } from '../scope.ts' import { SessionManager } from './manager.ts' import type { SessionRemotes } from './remotes.ts' import type { SessionListPhase, SessionSearchResultItem, SubagentCatalogSnapshot } from './manager.ts' -import { SessionProvideChannel } from './provide.ts' import type { Session } from './session.ts' /** Session list row projected from the host list RPC plus live stream increments. */ @@ -70,7 +71,7 @@ export interface SessionSummary { /** * Session list store shape. `current` rides the same snapshot (arbitrated: * the single useSessions standard hook reads list and selection together — - * sidebar highlighting and SessionProvider share one fact source). + * sidebar highlighting and current-session consumers share one fact source). */ export interface SessionListState { /** Host-list order; addressed breadcrumb-only rows are excluded. */ @@ -130,18 +131,20 @@ export class SessionForkError extends Error { } } -/** Session assembly handle for SessionProvider/inject factories (identity-stable per session). */ +/** Identity-stable logical binding for one materialized Client Session. */ export interface SessionBinding { readonly sessionId: SessionId /** The outward session face only — feature code never sees the concrete class. */ readonly session: SessionFace + /** Contiguous event window reserved for Conversation assembly. */ + readonly eventSource: SessionEventSource readonly ctx: AgentContext } -// Scope primitives live in ../agents/scope.ts (the client mirror of host +// Scope primitives live in ../scope.ts (the client mirror of host // dsh-scope, keyed by Agent identity); re-exported here so existing // consumers keep their import site. -export { scopeOf } from '../agents/scope.ts' +export { scopeOf } from '../scope.ts' /** * Workspace display title of a session cwd: the path's last non-empty @@ -194,34 +197,10 @@ interface ScopeRecord { binding: SessionBinding /** The concrete Session for runtime-internal entry points (staging open()); the binding carries only the outward face. */ session: Session - /** Render-layer standard-props bundle (identity-stable per scope; the renderer's per-info caches key off it). */ - provideInfo: SessionProvideInfo -} - -/** One plugin's per-session standard-props contribution (see {@link SessionRuntime.provide}). */ -export interface SessionProvideContribution { - /** Bare observable sources, keyed by hook base name ('input' → useInput). */ - hooks?: Record> - /** Stable plain members (action callbacks etc.), spread into standard props verbatim. */ - props?: Record -} - -/** - * Static declaration plus per-session resolver for one standard-kit - * contribution. The declared names let the renderer construct the same hook - * and prop surface while no session is current. - */ -export interface SessionProvideDescriptor { - /** Hook base names (`input` becomes `useInput`). */ - hooks?: readonly string[] - /** Plain standard-prop names. */ - props?: readonly string[] - /** Resolve every declared member for one definite session. */ - resolve(binding: SessionBinding): SessionProvideContribution } /** Root sessions service: list store, current selection, object-layer manager, scope tree, bindings, and breadcrumb routes. */ -export class SessionRuntime implements ISessions { +export class ClientSessions implements ISessions { /** * The wire schema's own result bound, re-exposed for presentation plugins as * injected data. Not per-connection state: the `session.search` response @@ -233,18 +212,10 @@ export class SessionRuntime implements ISessions { readonly list: SnapshotStore /** The object-layer instance cluster and frame dispatch entry. */ private readonly manager: SessionManager - /** - * Atomic current-session provide projection: selection changes and - * provider-roster changes publish through this one source (the renderer - * host's `sessions.provide` feed), so a roster change under a stable - * current id republishes the bundle instead of stranding mounted entries. - */ - readonly currentProvideInfo: HostObservable - /** * Persisted selection cell (the durable half of `list.current`). Private on * purpose: reads go through the list snapshot; writes through {@link - * SessionRuntime.open} / {@link SessionRuntime.clear}. Projection + * ClientSessions.open} / {@link ClientSessions.clear}. Projection * validates it against the live list instead of destructively pruning, so a * selection survives transient list states (reconnect re-pull) and * resurfaces when its session returns. @@ -252,8 +223,8 @@ export class SessionRuntime implements ISessions { private readonly selection: SnapshotStore private readonly scopes = new Map() - /** The provide channel (roster, materialization rules, current projection) — shared with the test runtime's double. */ - private readonly provideChannel: SessionProvideChannel + /** In-flight scope drops remain here after records leave `scopes`, so root disposal can await quiescence. */ + private readonly scopeDrops = new Set>() /** * The staged session id — follows `list.current` exactly, holding its last * defined value across masked gaps (a transiently absent selection blanks @@ -263,38 +234,26 @@ export class SessionRuntime implements ISessions { private watched: SessionId | undefined /** Removed-while-staged sessions whose teardown waits for the stage to move away. */ private readonly deferredRemovals = new Set() - /** Scope and journal teardowns started by synchronous list projection. */ - private readonly scopeDisposals = new Set>() /** * @param ctx - client root context (scope fibers mount under it). * @param api - wire client shared with every Session. * @param remote - generated Remote namespaces shared with every Session. - * @param conversationRuntime - same-pass registry instances, when runtime apply owns them. */ constructor( private readonly rootCtx: Context, api: IApiClient, remote: SessionRemotes, - conversationRuntime?: ConversationRuntime, ) { this.selection = createSnapshotStore( {}, { persist: { name: 'dsh.sessions.current' } }) const restored = this.selection.getSnapshot() - const conversationEvents = rootCtx.get('conversationEvents') - const conversationViews = rootCtx.get('conversationViews') - const conversation = conversationRuntime ?? ( - conversationEvents === undefined || conversationViews === undefined - ? undefined - : { events: conversationEvents, views: conversationViews } - ) this.manager = new SessionManager( api, remote, restored.sessionId, restored.subagentAddress, - conversation, ) this.list = createSnapshotStore({ ids: [], byId: {}, current: undefined, phase: 'pending', @@ -302,74 +261,32 @@ export class SessionRuntime implements ISessions { }) // The manager owns wire truth; the store is its projection. Manager // notifications are already microtask-batched. - this.manager.subscribe(() => { this.projectList() }) + const disposeManagerProjection = this.manager.subscribe(() => { + this.projectList() + }) // Stage follower: every current write (open() and projection alike) // re-evaluates staging, so startup restore (persisted selection validated // by the projection) and reconnect resurfacing open their window with no // dedicated code path. Safe to run synchronously inside the store notify: // the follower writes no list state — session.open()'s synchronous prefix // touches only session-side state and its own microtask-batched notifier. - // The current-provide projection follows the same current writes. - this.list.subscribe(() => { + const disposeStageFollower = this.list.subscribe(() => { this.followCurrent() - this.provideChannel.publishCurrent() }) - this.provideChannel = new SessionProvideChannel({ - rebuildBundles: () => { - for (const record of this.scopes.values()) { - record.provideInfo = this.provideChannel.materializeInfo(record.binding) - } - }, - resolveCurrent: () => this.maybeProvideInfo(this.list.getSnapshot().current), - }) - this.currentProvideInfo = this.provideChannel.currentProvideInfo - let registryRebuildQueued = false - const scheduleRegistryRebuild = (): void => { - if (registryRebuildQueued) return - registryRebuildQueued = true - queueMicrotask(() => { - registryRebuildQueued = false - this.manager.rebuildConversationRegistry() - }) - } - if (conversation !== undefined) { - rootCtx.effect(() => { - const disposeEvents = conversation.events.subscribe(scheduleRegistryRebuild) - const disposeViews = conversation.views.subscribe(scheduleRegistryRebuild) - return () => { - disposeEvents() - disposeViews() - } - }, 'sessions: conversation registry rebuild') - } rootCtx.effect(() => async () => { - for (const [id, record] of this.scopes) { - this.scopes.delete(id) - this.deferredRemovals.delete(id) - this.dropScope(id, record) - } - await Promise.all(this.scopeDisposals) - }, 'sessions: scoped resources') + disposeStageFollower() + disposeManagerProjection() + const scopes = [...this.scopes] + this.scopes.clear() + this.deferredRemovals.clear() + this.watched = undefined + for (const [id, record] of scopes) this.startScopeDrop(id, record) + await this.drainScopeDrops() + await this.manager.dispose() + }, 'session-controller.client.sessions') rootCtx.reflect.provide('sessions', this, undefined) } - /** - * Register a per-session standard-props provider: every session-scope slot - * component receives the contributed members as standard props (`hooks` - * sources become `use` selector hooks on the render side; `props` - * spread verbatim). Contributions materialize lazily with the session's - * scope record and die with it. Registration order is resolution order; - * duplicate member names fail loud at materialization. - * @param descriptor - static member roster plus per-session resolver. - * @returns disposer removing the provider (already-materialized bundles keep their members until their scope drops). - */ - provide(descriptor: SessionProvideDescriptor): () => void { - // Scopes may already exist (boot order: the list lands and resolves - // scopes before later plugins register) — the channel rebuilds their - // bundles through the host hooks so every provider lands by first render. - return this.provideChannel.provide(descriptor) - } - /** * Select a listed or retained catalog-addressed session as current. * @param id - listed or addressed session id. @@ -397,7 +314,7 @@ export class SessionRuntime implements ISessions { } /** - * Inform the runtime whether a catalog menu is consuming membership updates. + * Inform the Session Controller whether a catalog menu is consuming membership updates. * @param parentSessionId - selected parent. * @param open - menu state. */ @@ -498,7 +415,7 @@ export class SessionRuntime implements ISessions { this.manager.handleSessionError(...args) } - /** Refresh Session and subagent catalogs after connection; opened journals resume independently. */ + /** Rebuild the Session baseline and every opened window after connection. */ handleConnected(): void { this.manager.handleConnected() } @@ -506,7 +423,7 @@ export class SessionRuntime implements ISessions { /** * Create a session on the host. Resolution guarantee: by the time the * promise resolves, the created session is in the list store and - * {@link SessionRuntime.binding} resolves it — callers (New Session + * {@link ClientSessions.binding} resolves it — callers (New Session * draft hand-off) may address the scope synchronously, without waiting a * notifier flush. The synchronous projection below makes this structural * rather than an accident of microtask ordering. @@ -523,7 +440,7 @@ export class SessionRuntime implements ISessions { /** * Fork a session from a completed-turn prefix of the source (same - * synchronous-addressability guarantee as {@link SessionRuntime.create}: + * synchronous-addressability guarantee as {@link ClientSessions.create}: * on resolution the child is in the list store and open() can target it). * @param opts - source session id, the optional event seq anchoring the * cut (the boundary is the first turn/end at or after it; an in-log @@ -601,7 +518,7 @@ export class SessionRuntime implements ISessions { * hop every scoped consumer (event listeners, per-session controllers) * takes from ctx-space into object-space (the client mirror of host * `agent.session`). Same service-method boundary as - * {@link SessionRuntime.scopeOf}. + * {@link ClientSessions.scopeOf}. * @param ctx - an Agent-scoped context. * @returns the session face, or undefined when the ctx is untagged or its scope was pruned. */ @@ -621,25 +538,6 @@ export class SessionRuntime implements ISessions { return this.resolve(id)?.binding } - /** - * Resolve one session's render-layer standard-props bundle (ctx never - * enters the render layer; the renderer subscribes to - * {@link SessionRuntime.currentProvideInfo}). Pure resolution — render-safe: - * no staging, no window side effects (StrictMode double-invokes and - * concurrent discarded passes must stay free). - */ - private provideInfo(id: string): SessionProvideInfo | undefined { - return this.resolve(id as SessionId)?.provideInfo - } - - /** - * Resolve the current-session-optional standard kit. Unknown or absent ids - * return the static no-session projection rather than removing hook props. - */ - private maybeProvideInfo(id: string | undefined): SessionMaybeProvideInfo { - return (id === undefined ? undefined : this.provideInfo(id)) ?? this.provideChannel.maybeInfo - } - /** * Move the stage to the list's current session: sweep teardowns deferred * behind the previous occupant and pull the new occupant's history window. @@ -686,14 +584,12 @@ export class SessionRuntime implements ISessions { // The Session owns its scoped dispatch point (host Agent.loopCtx mirror); // mint and bind are one step so a live scope record implies a bound actx. session.bindScope(ctx) - const binding: SessionBinding = { sessionId: id, session, ctx } + const binding: SessionBinding = { sessionId: id, session, eventSource: session.eventSource, ctx } const record: ScopeRecord = { fiber, ctx, binding, session, - // Sources are bare observables; React binds selector hooks at its own boundary. - provideInfo: this.provideChannel.materializeInfo(binding), } this.scopes.set(id, record) return record @@ -790,7 +686,22 @@ export class SessionRuntime implements ISessions { } this.scopes.delete(id) this.deferredRemovals.delete(id) - this.dropScope(id, record) + this.startScopeDrop(id, record) + } + } + + private startScopeDrop(id: SessionId, record: ScopeRecord): void { + const drop = this.dropScope(id, record) + this.scopeDrops.add(drop) + void drop.then( + () => { this.scopeDrops.delete(drop) }, + () => { this.scopeDrops.delete(drop) }, + ) + } + + private async drainScopeDrops(): Promise { + while (this.scopeDrops.size > 0) { + await Promise.allSettled([...this.scopeDrops]) } } @@ -798,29 +709,17 @@ export class SessionRuntime implements ISessions { * One teardown for the whole per-session axis: the scope * fiber (cascading every actx-registered effect: input shell, slash * controller, popup, plugin stores, listeners), the session-keyed slot - * stores, and the Session instance itself — the host session log is the + * registrations and the Session instance itself — the host session log is the * durable truth, a reopen lazily rebuilds and backfills via open(). */ - private dropScope(id: SessionId, record: ScopeRecord): void { + private async dropScope(id: SessionId, record: ScopeRecord): Promise { // Release the Session's dispatch point with the scope it belongs to (a // surviving instance — the live Intent — rebinds when resolve re-mints). record.session.unbindScope() - // Optional lookup: slots and sessions are sibling services with no - // declared dependency; a slots-less boot (object-layer tests) skips. - this.rootCtx.get('slots')?.pruneStoreScope(id) - this.trackScopeDisposal(id, 'scope fiber', record.fiber.dispose()) - this.trackScopeDisposal(id, 'journal', this.manager.drop(id)) - } - - /** Retain one asynchronous scope cleanup through runtime disposal and contain its failure. */ - private trackScopeDisposal(id: SessionId, part: string, task: void | Promise): void { - const tracked = Promise.resolve(task).catch((error: unknown) => { - this.rootCtx.logger.warn( - `client-runtime: Session ${JSON.stringify(id)} ${part} cleanup failed: ${error instanceof Error ? error.message : String(error)}`, - ) - }) - this.scopeDisposals.add(tracked) - void tracked.then(() => { this.scopeDisposals.delete(tracked) }) + await Promise.allSettled([ + record.fiber.dispose(), + this.manager.drop(id), + ]) } /** Run deferred teardowns whose session is no longer staged (called when the stage moves). */ @@ -842,7 +741,7 @@ export class SessionRuntime implements ISessions { * future teardown path cannot double-dispose. */ if (record !== undefined) { this.scopes.delete(id) - this.dropScope(id, record) + this.startScopeDrop(id, record) } } } diff --git a/packages/client/runtime/src/client/sessions/session.ts b/packages/api/session-controller/src/client/sessions/session.ts similarity index 79% rename from packages/client/runtime/src/client/sessions/session.ts rename to packages/api/session-controller/src/client/sessions/session.ts index 6a373a7e7a..e894517e59 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/api/session-controller/src/client/sessions/session.ts @@ -3,19 +3,19 @@ import type { Context } from '@deepseek-ai/cordis' import { randomUUID } from '@deepseek-ai/dsh-util-crypto' import type { AttachmentIdType, ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' -import type { SessionEvent } from '@deepseek-ai/dsh-session/types' import type { - ClientFailure, ClientResult, IApiClient, MessageId, PromptContentPart, QueueAction, - SessionId, SubagentAddress, -} from '@deepseek-ai/dsh-api-remotes/client' + IApiClient, SubagentAddress, +} from '@deepseek-ai/dsh-client-connection/client' +import type { MessageId } from '@deepseek-ai/dsh-llm/brand' +import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types' import { SessionEventStream, sessionStreamFailure, -} from '@deepseek-ai/dsh-api-session-controller/client' -import type { - SessionEventChange, -} from '@deepseek-ai/dsh-api-session-controller/client' +} from '../transport.ts' +import type { SessionJournalChange } from '../transport.ts' import type { + PromptContentPart, + QueueAction, SessionAddress, SessionControlFrame, SessionEventEntry, @@ -23,18 +23,14 @@ import type { SessionRequestId, SessionError, SessionToolView, -} from '@deepseek-ai/dsh-api-session-controller/types' -// Value import from the inline-safe wire layer (not the connection plugin): -// plugin-to-plugin value imports are a bundle purity error. -import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api' +} from '../../types.ts' +import type { ClientFailure, ClientResult } from '../contract/result.ts' +import { transportResult } from '../contract/result.ts' import type { SessionFace } from '../contract/session.ts' -import { ConversationNodeAssembler } from './conversation-assembler.ts' -import type { ConversationRuntime } from './conversation-assembler.ts' -import type { ConversationEventInput, ConversationPublication } from '../contract/conversation.ts' import type { - ChatSnapshot, ComposerPhase, ConversationSnapshot, OpenState, PromptError, -} from './conversation.ts' -import { EMPTY_CHAT_SNAPSHOT } from './conversation.ts' + OpenState, PromptError, SessionSnapshot, +} from '../contract/snapshot.ts' +import { MutableSessionEventSource } from '../contract/events.ts' import { Notifier } from './notifier.ts' import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol' import type { SessionRemotes } from './remotes.ts' @@ -67,15 +63,13 @@ export interface SessionOptions { * private store (bare object-layer construction). */ projections?: ProjectionValueStore - /** Runtime registries used by this Session-owned Conversation assembler. */ - conversation?: ConversationRuntime } /** - * Owns a session's event window, derived conversation state, and observable + * Owns a session's event window, lifecycle state, and observable * snapshot. React bindings remain outside this data layer. Features see only * the {@link SessionFace} slice (ISession verbs + the snapshot source); the - * remaining public members are manager/runtime entry points. + * remaining public members are Session Controller internals. */ export class Session implements SessionFace { // ---- Window and derived state (all private; the snapshot is the only read API) ---- @@ -94,8 +88,6 @@ export class Session implements SessionFace { private loadingOlder = false /** Authoritative stream-only inbox snapshot; pending work never hits history. */ private readonly queueMirror = new SessionQueueMirror() - /** Session-owned business Context engine over the contiguous raw window. */ - private readonly conversation: ConversationNodeAssembler private running = false private address: SubagentAddress | undefined private parentAvailable = false @@ -129,10 +121,12 @@ export class Session implements SessionFace { */ readonly projections: ProjectionValueStore - private snapshotCache: ConversationSnapshot + /** Contiguous history and live tail consumed by Conversation assembly. */ + readonly eventSource = new MutableSessionEventSource() + private snapshotCache: SessionSnapshot private readonly notifier: Notifier /** - * Agent-scoped cordis context, bound once by SessionRuntime when it + * Agent-scoped cordis context, bound once by ClientSessions when it * mints the scope (the client mirror of the host Agent's loopCtx). The * Session dispatches its own scoped events through it; undefined means * unbound (bare object-layer construction) or already pruned — both skip @@ -155,21 +149,14 @@ export class Session implements SessionFace { this.projections = options.projections ?? new ProjectionValueStore() this.address = options.address this.parentAvailable = options.parentAvailable ?? false - this.conversation = options.conversation === undefined - ? new ConversationNodeAssembler( - { entries: () => [], fallbackEntry: () => undefined }, - { entries: () => [] }, - ) - : new ConversationNodeAssembler(options.conversation.events, options.conversation.views) this.notifier = new Notifier(() => { - this.conversation.flush() this.snapshotCache = this.buildSnapshot() }) this.snapshotCache = this.buildSnapshot() } /** - * Bind the Agent-scoped context minted by SessionRuntime (single write; + * Bind the Agent-scoped context minted by ClientSessions (single write; * a second bind is a wiring error and throws). Direction stays one-way at * this binding boundary: consumers still reach the Session via `sessions.sessionOf`, * while the Session holds its own dispatch point (host Agent.loopCtx @@ -249,7 +236,7 @@ export class Session implements SessionFace { } } } catch (error) { - result = transportError(error) + result = transportResult(error) } if (!result.ok) { this.promptError = { op: 'send', error: result.error } @@ -290,7 +277,7 @@ export class Session implements SessionFace { const data = Uint8Array.from(binary, char => char.charCodeAt(0)) return { ok: true, value: { attachment: result.value.attachment, data } } } catch (error) { - return transportError(error) + return transportResult(error) } } @@ -299,7 +286,7 @@ export class Session implements SessionFace { try { return toSessionResult(await this.remote.session.updateQueue({ sessionId: this.sessionId, itemId, action })) } catch (error) { - return transportError(error) + return transportResult(error) } } @@ -333,7 +320,7 @@ export class Session implements SessionFace { ? (await this.api.subagents.interrupt(address)).result : toSessionResult(await this.remote.session.cancel({ sessionId: this.sessionId })) } catch (error) { - result = transportError(error) + result = transportResult(error) } if (!result.ok) { this.promptError = { op: 'stop', error: result.error } @@ -357,7 +344,7 @@ export class Session implements SessionFace { if (result.ok) this.projections.apply('title', result.value.title, result.value.seq) return result } catch (error) { - return transportError(error) + return transportResult(error) } } @@ -397,7 +384,7 @@ export class Session implements SessionFace { await events.prepend({ beforeSeq: this.baseSeq, maxMessages: PAGE_MESSAGES }) } catch (error) { if (sessionStreamFailure(error) === undefined) { - console.error('[web-runtime] loadOlder failed:', error) + console.error('[session-controller] loadOlder failed:', error) } } finally { this.loadingOlder = false @@ -406,8 +393,8 @@ export class Session implements SessionFace { } /** Rebuild an opened history source after address replacement. - * Invalidates any in-flight open first; queue and pending-interaction state belongs - * to the independently reconnecting control stream and remains untouched. */ + * Invalidates any in-flight open first; queue state belongs to the independently + * reconnecting control stream and remains untouched. */ async resync(): Promise { if (this.openState === 'cold') return // never opened: no window to rebuild (doOpen flips to 'loading' synchronously, so cold implies no in-flight open) this.openGeneration++ @@ -436,10 +423,10 @@ export class Session implements SessionFace { } /** - * Cached conversation snapshot (rebuilt lazily when dirty with no listeners). + * Cached Session snapshot (rebuilt lazily when dirty with no listeners). * @returns the cached reference (stable until the next flush). */ - getSnapshot(): ConversationSnapshot { + getSnapshot(): SessionSnapshot { this.notifier.ensureFresh() return this.snapshotCache } @@ -538,18 +525,13 @@ export class Session implements SessionFace { /** * Stop the Session's live Remote source. - * @returns when the active journal generation and consumer are quiescent. + * @returns when the Remote iterator has completed teardown. */ - dispose(): Promise { + async dispose(): Promise { this.openGeneration++ const events = this.events this.events = undefined - return events?.dispose() ?? Promise.resolve() - } - - /** Rebuild the current window after a low-frequency Definition or view registration change. */ - rebuildConversationRegistry(): void { - this.scheduleConversation(this.conversation.rebuildRegistry()) + await events?.dispose() } // ---- Private ---- @@ -584,7 +566,7 @@ export class Session implements SessionFace { } /** Apply one contiguous journal update already reconciled by the Remote stream. */ - private acceptEventChange(change: SessionEventChange): void { + private acceptEventChange(change: SessionJournalChange): void { switch (change.type) { case 'replace': this.installWindow(change.entries, change.hasMore, change.page.projections) @@ -592,50 +574,42 @@ export class Session implements SessionFace { case 'prepend': this.prependWindow(change.entries, change.hasMore) return - case 'append': { - const entry = conversationInput(change.entry) - this.scheduleConversation(this.appendLive(entry.event, entry.view)) - } + case 'append': + if (this.appendLive(change.entry)) this.notifier.markDirty() } } /** Replace the complete contiguous window and apply page-owned projection metadata. */ private installWindow(entries: readonly SessionEventEntry[], hasMore: boolean, projections?: ProjectionsBaseline): void { - const normalized = entries.map(conversationInput) - this.eventWindow = normalized.map(entry => entry.event) - this.views = normalized.map(entry => entry.view) + this.eventWindow = entries.map(entry => entry.event as SessionEvent) + this.views = entries.map(entry => entry.view) this.baseSeq = this.eventWindow[0]?.seq ?? 0 this.hasMore = hasMore if (this.eventWindow.some(event => event.type === 'turn/start')) this.firstPromptPendingTurn = false - this.conversation.replaceWindow(normalized, hasMore) if (projections !== undefined) this.projections.seed(projections) + this.eventSource.replace(entries, hasMore) this.notifier.markDirty() } /** Prepend one stream-validated history page. */ private prependWindow(entries: readonly SessionEventEntry[], hasMore: boolean): void { - const normalized = entries.map(conversationInput) - this.eventWindow = [...normalized.map(entry => entry.event), ...this.eventWindow] - this.views = [...normalized.map(entry => entry.view), ...this.views] + this.eventWindow = [...entries.map(entry => entry.event as SessionEvent), ...this.eventWindow] + this.views = [...entries.map(entry => entry.view), ...this.views] this.baseSeq = this.eventWindow[0]?.seq ?? 0 this.hasMore = hasMore - this.conversation.prepend(normalized, hasMore) + this.eventSource.prepend(entries, hasMore) } /** Append one stream-validated live event. */ - private appendLive(event: SessionEvent, view?: SessionToolView): ConversationPublication { + private appendLive(entry: SessionEventEntry): boolean { + const event = entry.event as SessionEvent this.eventWindow.push(event) - this.views.push(view) + this.views.push(entry.view) + const awaitingFirstTurn = this.firstPromptPendingTurn if (event.type === 'turn/start') this.firstPromptPendingTurn = false const queueChanged = this.queueMirror.acceptDurable(event) - const publication = this.conversation.append({ event, view }) - return queueChanged ? 'immediate' : publication - } - - /** Route assembler cadence into the Session's existing microtask/RAF notifier. */ - private scheduleConversation(publication: ConversationPublication): void { - if (publication === 'immediate') this.notifier.markDirty() - else if (publication === 'animation-frame') this.notifier.markFrameDirty() + this.eventSource.append(entry) + return queueChanged || awaitingFirstTurn !== this.firstPromptPendingTurn } /** Publish a terminal background failure only while this stream still owns the Session. */ @@ -650,29 +624,14 @@ export class Session implements SessionFace { this.notifier.markDirty() } - private buildSnapshot(): ConversationSnapshot { - const chat = (this.conversation.snapshot('chat') as ChatSnapshot | undefined) ?? EMPTY_CHAT_SNAPSHOT - const legacy = chat.legacy + private buildSnapshot(): SessionSnapshot { return { sessionId: this.sessionId, - views: this.conversation, - chat, - nodes: legacy.nodes, - turnTimings: legacy.turnTimings, - turnEnds: legacy.turnEnds, - partial: legacy.partial, - runningCalls: legacy.runningCalls, queue: this.queueMirror.snapshot(), running: this.running, subagent: this.address === undefined ? null : { address: this.address, parentAvailable: this.parentAvailable }, - composerPhase: derivePhase( - hasVisibleConversationContent(chat) - || (!this.blankBit && !this.firstPromptPendingTurn) - || this.running, - this.promptAttempted, - ), removed: this.removed, openState: this.openState, openError: this.openError, @@ -681,6 +640,8 @@ export class Session implements SessionFace { promptError: this.promptError, blank: this.blankBit, lastAgentError: this.lastAgentError, + promptAttempted: this.promptAttempted, + awaitingFirstTurn: this.firstPromptPendingTurn, } } @@ -691,45 +652,16 @@ export class Session implements SessionFace { } } -/** Convert one wire history row into the assembler's transport-neutral input. */ -function conversationInput(entry: SessionEventEntry): ConversationEventInput { - return { - event: entry.event as SessionEvent, - view: entry.view, - } -} - /** Convert a terminal Session stream failure to the Client error vocabulary. */ function openFailure(error: unknown): ClientFailure { const failure = sessionStreamFailure(error) if (failure !== undefined) return failure as SessionError - const folded = transportError(error) - /* v8 ignore next -- transportError never returns an ok result. */ - if (folded.ok) throw new Error('transportError returned an unexpected success') + const folded = transportResult(error) + /* v8 ignore next -- transportResult never returns an ok result. */ + if (folded.ok) throw new Error('transportResult returned an unexpected success') return folded.error } - /** Narrow a generated Session Remote failure to its service-owned error vocabulary. */ function toSessionResult(result: RemoteResult): ClientResult { return result.ok ? result : { ok: false, error: result.error as SessionError } } - -/** A generic command row alone remains control-plane content; every other visible Chat Node activates the conversation. */ -function hasVisibleConversationContent(chat: ChatSnapshot): boolean { - return chat.order.some(key => chat.nodes.get(key)?.kind !== 'command') -} - -/** - * The composerPhase judgment — the single site that knows the predicate - * (consumers switch on the result, never re-derive). A failed first prompt - * stays engaging until an authoritative accepted-turn, running, or pending - * signal arrives (retry semantics — see ComposerPhase). - * @param hasContent - authoritative non-blank activity beyond a pending first - * prompt, visible non-command Chat content, a running turn, or a pending interaction. - * @param promptAttempted - a prompt was initiated on this session object. - * @returns the derived phase. - */ -function derivePhase(hasContent: boolean, promptAttempted: boolean): ComposerPhase { - if (hasContent) return 'active' - return promptAttempted ? 'engaging' : 'blank' -} diff --git a/packages/client/runtime/src/client/sessions/subagent-lineage.ts b/packages/api/session-controller/src/client/sessions/subagent-lineage.ts similarity index 92% rename from packages/client/runtime/src/client/sessions/subagent-lineage.ts rename to packages/api/session-controller/src/client/sessions/subagent-lineage.ts index 518f45ab1e..14409cbe10 100644 --- a/packages/client/runtime/src/client/sessions/subagent-lineage.ts +++ b/packages/api/session-controller/src/client/sessions/subagent-lineage.ts @@ -2,9 +2,9 @@ * Pure subagent-lineage aggregation over the retained session-list mirror. * Ordinary forks terminate propagation so each visible session owns only its * uninterrupted subagent subtree. - * @module @deepseek-ai/dsh-client-runtime/client/sessions/subagent-lineage + * @module @deepseek-ai/dsh-api-session-controller/client/sessions/subagent-lineage */ -import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' import type { SessionSummary } from './service.ts' /** Descendant counts projected for one possible parent session. */ diff --git a/packages/client/runtime/src/client/time-zone.ts b/packages/api/session-controller/src/client/time-zone.ts similarity index 100% rename from packages/client/runtime/src/client/time-zone.ts rename to packages/api/session-controller/src/client/time-zone.ts diff --git a/packages/api/session-controller/src/client/transport.ts b/packages/api/session-controller/src/client/transport.ts new file mode 100644 index 0000000000..2b7452056c --- /dev/null +++ b/packages/api/session-controller/src/client/transport.ts @@ -0,0 +1,181 @@ +/** Session-specific adapters for Gateway-owned Remote stream lifecycles. */ + +import type {} from '@deepseek-ai/dsh-api-session-controller/remote' +import type { RemoteFailure } from '@deepseek-ai/dsh-typert-protocol' +import { + RemoteJournalStream, + RemoteSnapshotStream, + RemoteStreamCarrierError, + RemoteStreamError, + type ClientRemote, + type RemoteJournalChange, + type RemoteJournalFrame, +} from '@deepseek-ai/dsh-api-gateway/client' +import type { + SessionAddress, + SessionControlFrame, + SessionEventEntry, + SessionPage, + SessionPageRequest, +} from '../types.ts' + +export { + SESSION_SEARCH_RESULT_LIMIT, + SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS, +} from '../types.ts' + +/** Pagination fields bound to an already-addressed Session journal. */ +export type ClientSessionPageRequest = Omit + +/** Complete generated `ctx.remote.session` namespace. */ +export type SessionRemote = ClientRemote['session'] + +/** One complete publication from the Session journal stream. */ +export type SessionJournalChange = RemoteJournalChange + +type SessionControlBaselineFrame = Extract +type SessionControlDeltaFrame = Exclude + +/** Gateway-owned control snapshot stream configured for Session frames. */ +export type SessionControlStream = RemoteSnapshotStream< + SessionControlBaselineFrame, + SessionControlDeltaFrame +> + +type SessionStreamRemote = Pick + +/** Domain sinks used by the Host-wide Session control stream. */ +export interface SessionControlStreamOptions { + /** Apply a complete baseline or one later update. */ + readonly accept: (frame: SessionControlFrame) => void + /** Observe a retryable carrier loss before reconnection. */ + readonly carrierFailed?: (error: RemoteStreamCarrierError) => void + /** Publish a terminal business or protocol failure. */ + readonly failed: (error: unknown) => void +} + +/** Domain sinks used by one addressed Session event journal. */ +export interface SessionEventStreamOptions { + /** Apply one complete event-window change. */ + readonly publish: (change: SessionJournalChange) => void + /** Observe a retryable carrier loss before reconnection. */ + readonly carrierFailed?: (error: RemoteStreamCarrierError) => void + /** Publish a terminal stream, page, or protocol failure after opening. */ + readonly failed: (error: unknown) => void +} + +/** + * Create the Host-wide Session control snapshot stream. + * @param remote - generated Session namespace and Gateway stream factory. + * @param options - Session state destinations. + * @returns an unstarted stream owned by the Client Session runtime. + */ +export function createSessionControlStream( + remote: SessionStreamRemote, + options: SessionControlStreamOptions, +): SessionControlStream { + const stream = remote.$stream({ + name: 'session control stream', + open: signal => remote.session.control(signal), + ended: accepted => accepted + ? new RemoteStreamCarrierError('session control stream ended without a terminal result') + : new Error('session control stream ended before its opening snapshot'), + ...(options.carrierFailed === undefined ? {} : { carrierFailed: options.carrierFailed }), + }) + return new RemoteSnapshotStream(stream, { + name: 'session control stream', + isSnapshot: (frame): frame is SessionControlBaselineFrame => frame.type === 'baseline', + replace: options.accept, + update: options.accept, + failed: options.failed, + }) +} + +/** Gateway-owned event journal bound to one ordinary or direct-subagent Session address. */ +export class SessionEventStream extends RemoteJournalStream< + SessionPage, + SessionEventEntry, + number, + ClientSessionPageRequest +> { + /** + * @param remote - generated Session namespace and Gateway stream factory. + * @param address - durable ordinary-Session or direct-subagent address. + * @param options - Session event-window destinations. + */ + constructor( + private readonly remote: SessionStreamRemote, + private readonly address: SessionAddress, + options: SessionEventStreamOptions, + ) { + super(remote, { + name: 'session event stream', + emptyCursor: -1, + entries: page => page.events, + hasMore: page => page.hasMore, + cursor: entry => entry.event.seq, + compare: (left, right) => left - right, + follows: (left, right) => right === left + 1, + publish: options.publish, + ...(options.carrierFailed === undefined + ? {} + : { carrierFailed: options.carrierFailed }), + failed: options.failed, + }) + } + + /** @inheritdoc */ + protected override async * follow( + afterSeq: number | undefined, + signal: AbortSignal, + ): AsyncIterable> { + const request = afterSeq === undefined + ? { address: this.address } + : { address: this.address, afterSeq } + for await (const frame of this.remote.session.follow(request, signal)) { + if (frame.type === 'opened') { + yield frame + continue + } + const { type: _type, ...entry } = frame + yield { type: 'entry', entry } + } + } + + /** @inheritdoc */ + protected override async readPage( + request: ClientSessionPageRequest, + throughSeq: number, + signal: AbortSignal, + ): Promise { + const result = await this.remote.session.page( + { address: this.address, throughSeq, ...request }, + signal, + ) + if (!result.ok) { + throw new RemoteStreamError( + result.error.code, + result.error.message, + result.error.details, + ) + } + return result.value + } + + /** @inheritdoc */ + protected override repairRequest( + request: ClientSessionPageRequest, + ): ClientSessionPageRequest { + return request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages } + } +} + +/** + * Recover a Host Session failure from a Remote stream terminal error. + * @param error - value thrown while opening or consuming a Session stream. + * @returns the Host failure, or `undefined` for carrier and local failures. + */ +export function sessionStreamFailure(error: unknown): RemoteFailure | undefined { + if (!(error instanceof RemoteStreamError)) return undefined + return { code: error.code, message: error.message, details: error.details } +} diff --git a/packages/api/session-controller/tests/client-apply.client.spec.ts b/packages/api/session-controller/tests/client-apply.client.spec.ts new file mode 100644 index 0000000000..04508378f9 --- /dev/null +++ b/packages/api/session-controller/tests/client-apply.client.spec.ts @@ -0,0 +1,226 @@ +import { Context } from '@deepseek-ai/cordis' +import type { Fiber } from '@deepseek-ai/cordis' +import type { + ConnectionHandle, + HostDescription, +} from '@deepseek-ai/dsh-client-connection/client' +import { + RemoteStreamCarrierError, + RemoteStream, + type RemoteStreamOptions, +} from '@deepseek-ai/dsh-api-gateway/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import TypertRegistry from '@deepseek-ai/dsh-typert-registry' +import { afterEach, describe, expect, it, vi } from 'vitest' +import * as SessionClient from '../src/client/index.ts' +import { ClientSessions } from '../src/client/sessions/service.ts' +import { FakeApiClient, fakeRemote } from './fake-api.client.ts' + +const DESCRIPTION: HostDescription = { + version: 'fixture', + cwd: '/fixture', + attachedSessions: 0, + home: '/home/fixture', + canOpenPath: true, +} + +const sid = (value: string): SessionId => value as SessionId + +type RemoteListener = (...args: never[]) => void + +interface Bench { + readonly ctx: Context + readonly api: FakeApiClient + readonly fiber: Fiber + readonly sessions: ClientSessions + dispatch(event: string, ...args: unknown[]): void + publishHost(description: HostDescription | undefined): void +} + +const contexts = new Set() + +afterEach(async () => { + vi.restoreAllMocks() + await Promise.all([...contexts].map(async (ctx) => { await ctx.fiber.dispose() })) + contexts.clear() +}) + +async function mount(initialHost?: HostDescription): Promise { + const ctx = new Context() + contexts.add(ctx) + await ctx.plugin(TypertRegistry) + const api = new FakeApiClient() + const remote = fakeRemote(api) + const listeners = new Map>() + const hostListeners = new Set<() => void>() + let host = initialHost + const connection: ConnectionHandle = { + api, + isLoopback: true, + hostDescription: { + getSnapshot: () => host, + subscribe: (listener) => { + hostListeners.add(listener) + return () => { hostListeners.delete(listener) } + }, + }, + rpc: { + call: () => Promise.reject(new Error('unexpected generic RPC call')), + }, + registerGenerationSource: () => () => {}, + start: () => ({ stop: () => {} }), + } + ctx.reflect.provide('connection', connection) + ctx.reflect.provide('remote', { + ...remote, + $stream: (options: RemoteStreamOptions) => ( + new RemoteStream(connection, options) + ), + $on: (event: string, listener: RemoteListener) => { + const eventListeners = listeners.get(event) ?? new Set() + eventListeners.add(listener) + listeners.set(event, eventListeners) + return () => { eventListeners.delete(listener) } + }, + }) + ctx.reflect.provide('remote.commands', remote.commands) + ctx.reflect.provide('remote.session', remote.session) + const fiber = ctx.plugin(SessionClient) + await fiber + const sessions = SessionClient.resolveClientSessions(ctx) as ClientSessions + return { + ctx, + api, + fiber, + sessions, + dispatch: (event, ...args) => { + for (const listener of listeners.get(event) ?? []) listener(...args as never[]) + }, + publishHost: (description) => { + host = description + for (const listener of [...hostListeners]) listener() + }, + } +} + +async function flush(): Promise { + for (let index = 0; index < 12; index++) await Promise.resolve() +} + +describe('Session Controller Client apply', () => { + it('requires the installed Session service at the resolver boundary', () => { + const ctx = new Context() + contexts.add(ctx) + + expect(() => SessionClient.resolveClientSessions(ctx)) + .toThrow('session-controller: Client sessions service unavailable') + }) + + it('routes Session Remote Events and connection generations into the object layer', async () => { + const connected = vi.spyOn(ClientSessions.prototype, 'handleConnected') + const error = vi.spyOn(ClientSessions.prototype, 'handleSessionError') + const bench = await mount() + expect(connected).not.toHaveBeenCalled() + + bench.dispatch('api-session/added', { + sessionId: sid('session-1'), + updatedAt: 1, + running: false, + blank: true, + }) + await flush() + expect(bench.sessions.list.getSnapshot().byId[sid('session-1')]).toMatchObject({ + running: false, + updatedAt: 1, + }) + + bench.dispatch('api-session/status', sid('session-1'), true) + bench.dispatch('api-session/activity', sid('session-1'), 9) + bench.dispatch('api-session/error', sid('session-1'), 'agent failed') + await flush() + expect(bench.sessions.list.getSnapshot().byId[sid('session-1')]).toMatchObject({ + running: true, + updatedAt: 9, + }) + expect(error).toHaveBeenCalledWith(sid('session-1'), 'agent failed') + + bench.dispatch('api-session/removed', sid('session-1')) + await flush() + expect(bench.sessions.list.getSnapshot().byId[sid('session-1')]).toBeUndefined() + + bench.ctx.emit('connection/reset') + expect(connected).toHaveBeenCalledOnce() + }) + + it('accepts the control baseline, retries a carrier generation, and reports terminal protocol failure', async () => { + const accept = vi.spyOn(ClientSessions.prototype, 'handleControlFrame') + const logged = vi.spyOn(console, 'error').mockImplementation(() => {}) + const bench = await mount(DESCRIPTION) + await flush() + + expect(accept).toHaveBeenCalledWith({ + type: 'baseline', + value: { queues: {}, jobs: {}, projections: {} }, + }) + + bench.api.failStreams(new RemoteStreamCarrierError('generation lost')) + await flush() + expect(accept.mock.calls.filter(([frame]) => frame.type === 'baseline')).toHaveLength(2) + + bench.api.pushControl({ type: 'baseline', value: bench.api.controlBaseline } as never) + await vi.waitFor(() => { + expect(logged).toHaveBeenCalledWith( + '[session-controller] control stream failed:', + expect.objectContaining({ message: 'session control stream emitted more than one opening snapshot' }), + ) + }) + }) + + it('materializes Host-addressed Agent scopes before the Session list arrives', async () => { + const bench = await mount() + const adapter = bench.ctx.typert.contexts.getClient('agent') + const first = adapter?.resolve(sid('agent-early')) + + expect(first).toBeDefined() + expect(bench.sessions.scopeOf(first as Context)).toBe(sid('agent-early')) + expect(adapter?.resolve(sid('agent-early'))).toBe(first) + }) + + it('projects Agent Context identity in both directions and withdraws the adapter on disposal', async () => { + const bench = await mount(DESCRIPTION) + await flush() + expect(bench.sessions.list.getSnapshot().phase).toBe('ready') + + bench.dispatch('api-session/added', { + sessionId: sid('agent-1'), + updatedAt: 1, + running: false, + blank: true, + }) + await flush() + const scoped = bench.sessions.scope(sid('agent-1')) + const adapter = bench.ctx.typert.contexts.getClient('agent') + expect(scoped).toBeDefined() + expect(adapter?.identity(bench.ctx)).toBeUndefined() + expect(adapter?.identity(scoped!)).toBe(sid('agent-1')) + expect(adapter?.resolve(sid('agent-1'))).toBe(scoped) + + await bench.fiber.dispose() + expect(bench.ctx.typert.contexts.getClient('agent')).toBeUndefined() + }) + + it('waits for a Host generation before retrying the control stream', async () => { + const accept = vi.spyOn(ClientSessions.prototype, 'handleControlFrame') + const bench = await mount() + await flush() + expect(accept.mock.calls.filter(([frame]) => frame.type === 'baseline')).toHaveLength(1) + + bench.api.failStreams(new RemoteStreamCarrierError('offline')) + await flush() + expect(accept.mock.calls.filter(([frame]) => frame.type === 'baseline')).toHaveLength(1) + + bench.publishHost(DESCRIPTION) + await flush() + expect(accept.mock.calls.filter(([frame]) => frame.type === 'baseline')).toHaveLength(2) + }) +}) diff --git a/packages/api/session-controller/tests/client-contract.client.spec.ts b/packages/api/session-controller/tests/client-contract.client.spec.ts new file mode 100644 index 0000000000..18919e66e9 --- /dev/null +++ b/packages/api/session-controller/tests/client-contract.client.spec.ts @@ -0,0 +1,67 @@ +import type { SessionEventEntry } from '@deepseek-ai/dsh-api-session-controller/types' +import { describe, expect, it, vi } from 'vitest' +import { MutableSessionEventSource } from '../src/client/contract/events.ts' +import { transportResult } from '../src/client/contract/result.ts' + +function entry(seq: number): SessionEventEntry { + return { + event: { + type: 'fixture/event', + seq, + time: seq, + data: { seq }, + ignorable: true, + }, + } +} + +describe('Client Session contracts', () => { + it('publishes exact replace, prepend, and append event-window changes', () => { + const feed = new MutableSessionEventSource() + const listener = vi.fn() + const dispose = feed.subscribe(listener) + const first = entry(1) + const older = entry(0) + const live = entry(2) + + feed.replace([first], true) + expect(feed.getSnapshot()).toEqual({ + entries: [first], + hasMore: true, + revision: 1, + change: { kind: 'replace', entries: [first] }, + }) + + feed.prepend([older], false) + expect(feed.getSnapshot()).toEqual({ + entries: [older, first], + hasMore: false, + revision: 2, + change: { kind: 'prepend', entries: [older] }, + }) + + feed.append(live) + expect(feed.getSnapshot()).toEqual({ + entries: [older, first, live], + hasMore: false, + revision: 3, + change: { kind: 'append', entries: [live] }, + }) + expect(listener).toHaveBeenCalledTimes(3) + + dispose() + feed.append(entry(3)) + expect(listener).toHaveBeenCalledTimes(3) + }) + + it('folds Error and non-Error carrier rejections into Client failures', () => { + expect(transportResult(new Error('transport unavailable'))).toEqual({ + ok: false, + error: { code: 'internal', message: 'transport unavailable', details: {} }, + }) + expect(transportResult(404)).toEqual({ + ok: false, + error: { code: 'internal', message: '404', details: {} }, + }) + }) +}) diff --git a/packages/client/runtime/tests/event-script.client.ts b/packages/api/session-controller/tests/event-script.client.ts similarity index 98% rename from packages/client/runtime/tests/event-script.client.ts rename to packages/api/session-controller/tests/event-script.client.ts index a024af0ab7..0781621296 100644 --- a/packages/client/runtime/tests/event-script.client.ts +++ b/packages/api/session-controller/tests/event-script.client.ts @@ -1,4 +1,6 @@ -import { createUserMessage, createMessage, createToolResultMessage, CallId } from '@deepseek-ai/dsh-llm' +import { + CallId, createMessage, createToolResultMessage, createUserMessage, +} from '@deepseek-ai/dsh-llm' // Minimal SessionEvent builders for orchestration tests (shape mirrors what the // host emits; only the fields the object layer reads). import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' diff --git a/packages/client/runtime/tests/fake-api.client.ts b/packages/api/session-controller/tests/fake-api.client.ts similarity index 93% rename from packages/client/runtime/tests/fake-api.client.ts rename to packages/api/session-controller/tests/fake-api.client.ts index 45cc2e30cd..6c82ab0352 100644 --- a/packages/client/runtime/tests/fake-api.client.ts +++ b/packages/api/session-controller/tests/fake-api.client.ts @@ -1,9 +1,9 @@ // Test-local programmable IApiClient fake (NOT the fixture: fixture is a demo // data source on a real clock; behavior tests need per-case responses and -// deferred-controlled timing). Streams are hand pumps: pushFollow/pushControl/pushWorkspace. +// deferred-controlled timing). Session streams are hand pumps: pushFollow/pushControl. import type { - IApiClient, ModelSelection, - RpcError, RpcResponse, SessionId, SessionModels, SessionSearchItem, SkillEntry, + IApiClient, + RpcError, RpcResponse, SessionId, SessionSearchItem, SkillEntry, WorkspaceId, WorkspaceView, } from '@deepseek-ai/dsh-api-remotes/client' import type { @@ -12,11 +12,14 @@ import type { SessionControlFrame, SessionFollowFrame, SessionFollowRequest, + SessionModels, SessionPage, SessionPageRequest, + SessionSelectModelRequest, + SessionSelectModelValue, } from '@deepseek-ai/dsh-api-session-controller/types' import type { WorkspaceRemote } from '@deepseek-ai/dsh-api-workspace-controller/client' -import type { WorkspaceError, WorkspaceFollowFrame } from '@deepseek-ai/dsh-api-workspace-controller/types' +import type { WorkspaceFollowFrame } from '@deepseek-ai/dsh-api-workspace-controller/types' import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol' import { RemoteStream, @@ -85,15 +88,10 @@ export function err(error: RpcError): RpcResponse { } /** Successful generated Remote result for programmable domain fakes. */ -export function remoteOk(value: T): RemoteResult { +function remoteOk(value: T): RemoteResult { return { ok: true, value } } -/** Workspace business failure returned by a generated Remote fake. */ -export function workspaceErr(error: WorkspaceError): RemoteResult { - return { ok: false, error } -} - type ValueStreamItem = | { kind: 'frame'; value: F; delivered?: () => void } | { kind: 'end' } @@ -130,26 +128,28 @@ export class FakeApiClient implements IApiClient { onSearch: (payload: unknown) => Promise> = () => Promise.resolve(ok({ items: [], hasMore: false })) onCreate: (payload: unknown) => Promise> = () => Promise.resolve(ok({ sessionId: 'fk-new' as SessionId })) - readonly defaultModel: ModelSelection = { provider: 'deepseek-official', model: 'deepseek-v4-flash' } + onModels: (payload: unknown) => Promise> = () => Promise.resolve(ok({ + current: { provider: 'fixture', model: 'fixture' }, + routable: true, + groups: [], + failures: [], + })) + onSelectModel: (payload: SessionSelectModelRequest) => Promise> = + payload => Promise.resolve(ok({ + selected: { + provider: payload.provider, + model: payload.model, + ...(payload.reasoningEffort === undefined + ? {} + : { reasoningEffort: payload.reasoningEffort }), + }, + })) onRename: (payload: unknown) => Promise> = () => Promise.resolve(ok({ title: 'fk-renamed', seq: 0 })) onFork: (payload: unknown) => Promise> = () => Promise.resolve(ok({ sessionId: 'fk-fork' as SessionId })) onHistory: (payload: { sessionId: SessionId; throughSeq?: number; beforeSeq?: number; maxMessages?: number }) => Promise> = () => Promise.resolve(ok({ events: [], hasMore: false })) - onModels: (payload: unknown) => Promise> = () => Promise.resolve(ok({ - current: this.defaultModel, - routable: true, - groups: [{ - id: 'deepseek-official', - name: 'DeepSeek', - models: [{ id: 'deepseek-v4-flash', name: 'DeepSeek V4 Flash' }], - }], - failures: [], - })) - onSelectModel: (payload: { provider: string; model: string }) => - Promise> = - payload => Promise.resolve(ok({ selected: { provider: payload.provider, model: payload.model } })) onPrompt: (payload: unknown) => Promise> = () => Promise.resolve(ok({ accepted: true as const })) onAttachment: (payload: unknown) => Promise> = () => Promise.resolve(ok({ attachment: { attachmentId: 'a' as never, mediaType: 'image/png', bytes: 1, width: 1, height: 1 }, data: 'AA==' })) @@ -303,7 +303,6 @@ export class FakeApiClient implements IApiClient { new RemoteStream(AVAILABLE_STREAM_CONNECTION, options) ), commands: { - list: () => Promise.resolve({ ok: true, value: [] }), execute: () => Promise.resolve({ ok: true, value: undefined }), }, session: { @@ -314,7 +313,11 @@ export class FakeApiClient implements IApiClient { }, create: payload => this.remoteResult('session.create', payload, this.onCreate(payload)), models: payload => this.remoteResult('session.models', payload, this.onModels(payload)), - selectModel: payload => this.remoteResult('session.selectModel', payload, this.onSelectModel(payload)), + selectModel: payload => this.remoteResult( + 'session.selectModel', + payload, + this.onSelectModel(payload), + ), rename: payload => this.remoteResult('session.rename', payload, this.onRename(payload)), fork: payload => this.remoteResult('session.fork', payload, this.onFork(payload)), prompt: payload => this.remoteResult('session.prompt', payload, this.onPrompt(payload)), @@ -464,19 +467,24 @@ export class FakeApiClient implements IApiClient { const sessionId = addressSessionId(request.address) this.followStarts.push(sessionId) const key = addressKey(request.address) - const initialPage = this.onHistory({ sessionId, maxMessages: 50 }) - this.openingPages.set(key, initialPage) + const initialPage = this.followCursor === undefined + ? this.onHistory({ sessionId, maxMessages: 50 }) + : undefined + if (initialPage !== undefined) this.openingPages.set(key, initialPage) const conns = this.followConns.get(sessionId) ?? [] if (!this.followConns.has(sessionId)) this.followConns.set(sessionId, conns) const stream = this.openValueStream(conns, signal) try { - const page = (await initialPage).result - const cursor = this.followCursor ?? (page.ok ? page.value.events.at(-1)?.event.seq ?? -1 : -1) + const page = initialPage === undefined ? undefined : (await initialPage).result + const cursor = this.followCursor + ?? (page?.ok ? page.value.events.at(-1)?.event.seq ?? -1 : -1) yield { type: 'opened', cursor } yield* stream.values } finally { stream.dispose() - this.openingPages.delete(key) + if (initialPage !== undefined && this.openingPages.get(key) === initialPage) { + this.openingPages.delete(key) + } } } diff --git a/packages/client/runtime/tests/lineage.client.spec.ts b/packages/api/session-controller/tests/lineage.client.spec.ts similarity index 98% rename from packages/client/runtime/tests/lineage.client.spec.ts rename to packages/api/session-controller/tests/lineage.client.spec.ts index d0c657c46b..b15b89e61b 100644 --- a/packages/client/runtime/tests/lineage.client.spec.ts +++ b/packages/api/session-controller/tests/lineage.client.spec.ts @@ -12,7 +12,7 @@ const s = (id: string, updatedAt: number, parent?: string): SessionSummary => ({ ...(parent !== undefined ? { parentSessionId: parent as SessionId } : {}), }) -describe('flattenLineage', () => { +describe('Session lineage flattening', () => { it('keeps established root and sibling order while expanding children DFS with depth', () => { const out = flattenLineage([ s('old-root', 10), diff --git a/packages/client/runtime/tests/manager.client.spec.ts b/packages/api/session-controller/tests/manager.client.spec.ts similarity index 99% rename from packages/client/runtime/tests/manager.client.spec.ts rename to packages/api/session-controller/tests/manager.client.spec.ts index 45acd7c8fd..26443dbb5f 100644 --- a/packages/client/runtime/tests/manager.client.spec.ts +++ b/packages/api/session-controller/tests/manager.client.spec.ts @@ -1,6 +1,6 @@ /** * SessionManager orchestration: lazy resident instances, list lifecycle, host - * frame routing, and the pending-frame buffer for uninstantiated sessions. + * frame routing, and control baselines for uninstantiated sessions. */ import { describe, expect, it, vi } from 'vitest' @@ -32,7 +32,7 @@ function makeManager(): SessionManager { return new SessionManager(api, fakeRemote(api)) } -describe('instances', () => { +describe('SessionManager instances', () => { it('lazily builds one resident instance per id and syncs the running bit from the list', async () => { const api = new FakeApiClient() api.onList = () => Promise.resolve(ok({ items: [summary(S1, { running: true })] as never[] })) @@ -710,11 +710,9 @@ describe('remaining branches', () => { expect(notified).toBe(seen) }) - it('dispatches Host events to instantiated sessions', () => { + it('ignores Host status and error events for sessions without an instance', () => { const api = new FakeApiClient() const manager = new SessionManager(api, fakeRemote(api)) - manager.get(S1) - // status flip for an unknown session only touches summaries (no crash). manager.handleSessionStatus(S2, true) manager.handleSessionError(S2, '无实例') }) diff --git a/packages/client/runtime/tests/notifier.client.spec.ts b/packages/api/session-controller/tests/notifier.client.spec.ts similarity index 99% rename from packages/client/runtime/tests/notifier.client.spec.ts rename to packages/api/session-controller/tests/notifier.client.spec.ts index f12d063400..dc5ca2f4ff 100644 --- a/packages/client/runtime/tests/notifier.client.spec.ts +++ b/packages/api/session-controller/tests/notifier.client.spec.ts @@ -12,7 +12,7 @@ afterEach(() => { vi.unstubAllGlobals() }) -describe('Notifier', () => { +describe('Session notifier', () => { it('collapses N markDirty calls into one flush, rebuilding before notifying', async () => { const order: string[] = [] const notifier = new Notifier(() => order.push('rebuild')) diff --git a/packages/client/runtime/tests/projection-store.client.spec.ts b/packages/api/session-controller/tests/projection-store.client.spec.ts similarity index 99% rename from packages/client/runtime/tests/projection-store.client.spec.ts rename to packages/api/session-controller/tests/projection-store.client.spec.ts index 850c21b478..f12bd4475b 100644 --- a/packages/client/runtime/tests/projection-store.client.spec.ts +++ b/packages/api/session-controller/tests/projection-store.client.spec.ts @@ -25,7 +25,7 @@ declare module '@deepseek-ai/dsh-session-projection/types' { const SID = 'fk-s1' as SessionId -describe('ProjectionValueStore semantics', () => { +describe('Session projection value semantics', () => { it('reads undefined until a value lands (capability absence)', () => { const store = new ProjectionValueStore() expect(store.get('test/marks')).toBeUndefined() diff --git a/packages/client/runtime/tests/queue-store.client.spec.ts b/packages/api/session-controller/tests/queue-store.client.spec.ts similarity index 99% rename from packages/client/runtime/tests/queue-store.client.spec.ts rename to packages/api/session-controller/tests/queue-store.client.spec.ts index 496762f748..ed0a1568d1 100644 --- a/packages/client/runtime/tests/queue-store.client.spec.ts +++ b/packages/api/session-controller/tests/queue-store.client.spec.ts @@ -56,7 +56,7 @@ function makeManager(): SessionManager { return new SessionManager(api, fakeRemote(api)) } -describe('queue snapshot intake', () => { +describe('Session queue snapshot intake', () => { it('projects stable ids, flat previews, and complete text', () => { const session = makeSession() session.handleControlFrame(queueFrame([ diff --git a/packages/client/runtime/tests/scope.client.spec.ts b/packages/api/session-controller/tests/scope.client.spec.ts similarity index 97% rename from packages/client/runtime/tests/scope.client.spec.ts rename to packages/api/session-controller/tests/scope.client.spec.ts index 528c36131e..a0d7be3e7e 100644 --- a/packages/client/runtime/tests/scope.client.spec.ts +++ b/packages/api/session-controller/tests/scope.client.spec.ts @@ -9,7 +9,7 @@ import { Context } from '@deepseek-ai/cordis' import { describe, expect, it } from 'vitest' import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client' -import { createScope, scopeOf } from '../src/client/agents/scope.ts' +import { createScope, scopeOf } from '../src/client/scope.ts' const sid = (k: string): SessionId => k as SessionId diff --git a/packages/client/runtime/tests/session.client.spec.ts b/packages/api/session-controller/tests/session.client.spec.ts similarity index 66% rename from packages/client/runtime/tests/session.client.spec.ts rename to packages/api/session-controller/tests/session.client.spec.ts index fae04f68bd..17533bb585 100644 --- a/packages/client/runtime/tests/session.client.spec.ts +++ b/packages/api/session-controller/tests/session.client.spec.ts @@ -1,24 +1,11 @@ -/** - * Session orchestration: drive the object through contract calls and injected - * frames (open → prompt → stream → finalize → cancel → resync) and assert the - * ConversationSnapshot it settles into. Reference stability is asserted with - * toBe/not.toBe — it is the React.memo/uSES contract, equal-value output is not - * enough. - */ +/** Session object lifecycle, event-window transport, commands, and resync behavior. */ import { afterEach, describe, expect, it, vi } from 'vitest' import { RemoteStreamError } from '@deepseek-ai/dsh-api-gateway/client' import type { SessionEvent } from '@deepseek-ai/dsh-session/types' -import type {} from '@deepseek-ai/dsh-commands/types' import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client' import type { SessionToolView } from '@deepseek-ai/dsh-api-session-controller/types' -import { Session } from '../src/client/sessions/session.ts' -import type { - ChatConversationViewNode, ChatLocationNodeIndex, ChatNodeStore, ChatSnapshot, - ConversationEventInput, ConversationNode, ConversationNodeDefinition, - ConversationRuntime, ConversationSnapshot, ConversationTimelineSnapshot, - ConversationViewDefinition, -} from '../src/client/index.ts' +import { Session, type SessionOptions } from '../src/client/sessions/session.ts' import { FakeApiClient, deferred, err, fakeRemote, ok } from './fake-api.client.ts' import { entries, ev, plainTurn } from './event-script.client.ts' @@ -29,139 +16,11 @@ afterEach(() => { vi.unstubAllGlobals() }) -const EMPTY: readonly never[] = [] - -interface TestEventState extends ConversationEventInput {} - -class TestNodeStore implements ChatNodeStore { - private readonly nodes = new Map() - private cache: readonly ChatConversationViewNode[] = EMPTY - - get(key: string): ChatConversationViewNode | undefined { - return this.nodes.get(key) - } - - values(): readonly ChatConversationViewNode[] { - return this.cache - } - - replace(nodes: readonly ChatConversationViewNode[]): void { - this.nodes.clear() - for (const node of nodes) this.nodes.set(node.key, node) - this.cache = [...this.nodes.values()] - } - - upsert(nodes: readonly ChatConversationViewNode[]): void { - if (nodes.length === 0) return - for (const node of nodes) this.nodes.set(node.key, node) - this.cache = [...this.nodes.values()] - } -} - -const TEST_LOCATIONS: ChatLocationNodeIndex = { - getTurn: () => EMPTY, - getStep: () => EMPTY, -} - -function testLegacy( - nodes: readonly ChatConversationViewNode[], - timeline: ConversationTimelineSnapshot, -): ChatSnapshot['legacy'] { - const legacyNodes = nodes.flatMap((node): ConversationNode[] => { - const event = (node.data as TestEventState).event - if (event.type === 'user/message') return [{ kind: 'user', seq: event.seq } as ConversationNode] - if (event.type === 'assistant/message') return [{ kind: 'assistant', seq: event.seq } as ConversationNode] - return [] - }) - const turnTimings = new Map() - const turnEnds = new Map() - for (const turn of timeline.turns.values()) { - if (turn.start !== undefined) { - turnTimings.set(turn.turn, turn.end === undefined - ? { startTime: turn.start.time } - : { startTime: turn.start.time, endTime: turn.end.time }) - } - if (turn.end !== undefined) turnEnds.set(turn.turn, turn.end.seq) - } - return { nodes: legacyNodes, turnTimings, turnEnds, partial: null, runningCalls: EMPTY } -} - -function testViewDefinition(): ConversationViewDefinition { - return { - target: 'chat', - create: () => { - const store = new TestNodeStore() - let current: ChatSnapshot = { - order: EMPTY, - nodes: store, - locations: TEST_LOCATIONS, - timeline: { turnOrder: EMPTY, turns: new Map() }, - legacy: testLegacy(EMPTY, { turnOrder: EMPTY, turns: new Map() }), - } - const build = (timeline: ConversationTimelineSnapshot): ChatSnapshot => { - const nodes = [...store.values()].sort((left, right) => left.anchorSeq - right.anchorSeq) - current = { - order: nodes.map(node => node.key), - nodes: store, - locations: TEST_LOCATIONS, - timeline, - legacy: testLegacy(nodes, timeline), - } - return current - } - return { - empty: current, - replace: ({ nodes, timeline }) => { - store.replace(nodes) - return build(timeline) - }, - apply: ({ upserts, timeline }) => { - store.upsert(upserts) - return build(timeline) - }, - } - }, - } -} - -const TEST_EVENT_DEFINITION: ConversationNodeDefinition = { - kind: 'runtime-test-event', - target: 'chat', - match: event => ({ id: String(event.seq), role: 'start' }), - start: (_context, match) => ({ event: match.event, view: match.view }), - update: context => context.state, - publication: match => match.event.type === 'assistant/chunk' ? 'animation-frame' : 'immediate', - buildViewNode: (context) => { - if (context.state === undefined || context.start === undefined) return null - return { - key: context.key, - kind: context.start.event.type === 'command/run' && context.start.event.data.name === 'goal' - ? 'command-input' - : context.start.event.type === 'command/run' || context.start.event.type === 'command/done' - ? 'command' - : 'runtime-test-event', - id: context.id, - target: 'chat', - anchorSeq: context.start.event.seq, - location: context.start.location, - visibility: 'visible', - data: context.state, - } - }, -} - -const TEST_CONVERSATION: ConversationRuntime = { - events: { - entries: () => [TEST_EVENT_DEFINITION], - fallbackEntry: () => undefined, - } as unknown as ConversationRuntime['events'], - views: { - entries: () => [testViewDefinition()], - } as unknown as ConversationRuntime['views'], -} - -function makeSession(api = new FakeApiClient()): { api: FakeApiClient; session: Session } { - return { api, session: new Session(SID, api, fakeRemote(api), { conversation: TEST_CONVERSATION }) } +function makeSession( + api = new FakeApiClient(), + options: SessionOptions = {}, +): { api: FakeApiClient; session: Session } { + return { api, session: new Session(SID, api, fakeRemote(api), options) } } function follow( @@ -176,12 +35,12 @@ function follow( }) } -function chatEvents(snapshot: ConversationSnapshot): readonly TestEventState[] { - return snapshot.chat.order.map(key => snapshot.chat.nodes.get(key)?.data as TestEventState) +function windowEntries(session: Session) { + return session.eventSource.getSnapshot().entries } -function chatSeqs(snapshot: ConversationSnapshot): number[] { - return chatEvents(snapshot).map(item => item.event.seq) +function eventSeqs(session: Session): number[] { + return windowEntries(session).map(entry => entry.event.seq) } function histResponse(events: SessionEvent[], hasMore = false) { @@ -189,13 +48,13 @@ function histResponse(events: SessionEvent[], hasMore = false) { return Promise.resolve(ok({ events: entries(events) as never[], hasMore })) } -describe('open', () => { +describe('Session open', () => { it('keeps a bare Session blank until an authoritative lifecycle signal arrives', () => { const { session } = makeSession() - expect(session.getSnapshot()).toMatchObject({ blank: true, composerPhase: 'blank' }) + expect(session.getSnapshot()).toMatchObject({ blank: true, promptAttempted: false, running: false }) session.handleRunning(true) - expect(session.getSnapshot()).toMatchObject({ blank: false, composerPhase: 'active' }) + expect(session.getSnapshot()).toMatchObject({ blank: false, running: true }) }) it('installs the tail page: cold → loading → open with window and nodes in place', async () => { @@ -209,12 +68,8 @@ describe('open', () => { const snapshot = session.getSnapshot() expect(snapshot.openState).toBe('open') expect(snapshot.hasMore).toBe(true) - expect(snapshot.nodes.map(n => n.kind)).toEqual(['user', 'assistant']) - expect(snapshot.turnTimings.get(3)).toEqual({ - startTime: 1_700_000_000_010, - endTime: 1_700_000_000_015, - }) - expect(snapshot.turnEnds.get(3)).toBe(15) + expect(eventSeqs(session)).toEqual([10, 11, 12, 13, 14, 15]) + expect(session.eventSource.getSnapshot().change).toMatchObject({ kind: 'replace' }) }) it('is idempotent: concurrent opens share one history call, reopening when open is a no-op', async () => { @@ -258,9 +113,9 @@ describe('open', () => { modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' }, })) await Promise.all([opening, ...deliveries]) - const seqs = session.getSnapshot().nodes.map(n => n.seq) + const seqs = eventSeqs(session) // Overlapping seq-15 frame (== page tail turn/end) was dropped; 16 appended once. - expect(seqs).toEqual([11, 13, 16]) + expect(seqs).toEqual([10, 11, 12, 13, 14, 15, 16]) }) }) @@ -275,98 +130,21 @@ describe('live event path', () => { it('drops replayed frames at or below the window tail', async () => { const { api, session } = await opened() - const before = session.getSnapshot() + const before = session.eventSource.getSnapshot() await follow(api, ev.user(3, '重放')) - expect(session.getSnapshot().nodes).toEqual(before.nodes) + expect(session.eventSource.getSnapshot()).toBe(before) }) it('keeps the authoritative host blank bit across unrelated log events', async () => { const { api, session } = await opened([]) session.handleBlank(true) - expect(session.getSnapshot().composerPhase).toBe('blank') await Promise.all([ follow(api, ev.commandRun(0, 'cmd-perm', 'permission', ' danger-full-access')), follow(api, ev.commandDone(1, 'cmd-perm', 'success', 'preset danger-full-access')), ]) const snapshot = session.getSnapshot() - expect(chatSeqs(snapshot)).toEqual([0, 1]) - expect(snapshot.composerPhase).toBe('blank') - }) - - it('activates a fresh conversation for a command-input View Node without opening a model turn', async () => { - const { api, session } = await opened([]) - session.handleBlank(true) - await Promise.all([ - follow(api, ev.commandRun(0, 'cmd-goal', 'goal', ' ')), - follow(api, ev.commandDone(1, 'cmd-goal', 'success', 'No goal is currently set.')), - ]) - - expect(session.getSnapshot()).toMatchObject({ - blank: true, - composerPhase: 'active', - }) - expect(session.getSnapshot().chat.order.map( - key => session.getSnapshot().chat.nodes.get(key)?.kind, - )).toContain('command-input') - }) - - it('publishes animation-frame Definitions once per frame and lets an immediate event supersede the pending frame', async () => { - const frames: FrameRequestCallback[] = [] - vi.stubGlobal('requestAnimationFrame', (callback: FrameRequestCallback) => { - frames.push(callback) - return frames.length - }) - const { api, session } = await opened() - const published: number[][] = [] - session.subscribe(() => { - published.push(chatSeqs(session.getSnapshot())) - }) - await Promise.all([ - follow(api, ev.chunkStart(6, 1)), - follow(api, ev.chunkText(7, 1, '累')), - follow(api, ev.chunkText(8, 1, '计')), - ]) - expect(published).toEqual([]) - expect(frames).toHaveLength(1) - - frames.shift()!(0) - expect(published).toEqual([[0, 1, 2, 3, 4, 5, 6, 7, 8]]) - - await Promise.all([ - follow(api, ev.chunkText(9, 1, '完成')), - follow(api, ev.assistant(10, 1, '累计完成')), - ]) - await Promise.resolve() - expect(published).toEqual([ - [0, 1, 2, 3, 4, 5, 6, 7, 8], - [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10], - ]) - - frames.shift()!(0) - expect(published).toHaveLength(2) - }) - - it('publishes a timeline-only boundary even when no Definition claims the event', async () => { - const api = new FakeApiClient() - api.onHistory = () => histResponse([]) - const conversation: ConversationRuntime = { - events: { - entries: () => [], - fallbackEntry: () => undefined, - } as unknown as ConversationRuntime['events'], - views: { - entries: () => [testViewDefinition()], - } as unknown as ConversationRuntime['views'], - } - const session = new Session(SID, api, fakeRemote(api), { conversation }) - await session.open() - const snapshots: ConversationSnapshot[] = [] - session.subscribe(() => { snapshots.push(session.getSnapshot()) }) - - await follow(api, ev.turnStart(0, 1)) - - expect(snapshots).toHaveLength(1) - expect(snapshots[0]?.chat.timeline.turns.get(1)?.status).toBe('open') + expect(eventSeqs(session)).toEqual([0, 1]) + expect(snapshot.blank).toBe(true) }) it('repairs a seq gap by repulling the tail page instead of appending a hole', async () => { @@ -379,8 +157,9 @@ describe('live event path', () => { expect(api.callsOf('session.history').length).toBe(2) }) await vi.waitFor(() => { - const seqs = session.getSnapshot().nodes.map(n => n.seq) - expect(seqs).toEqual([1, 3, 7, 9]) // both turns' user/assistant, no hole, no duplicate 9 + expect(eventSeqs(session)).toEqual( + repaired.filter(event => event.seq <= 9).map(event => event.seq), + ) }) }) }) @@ -401,7 +180,7 @@ describe('paging', () => { { sessionId: SID, throughSeq: 11, beforeSeq: 6 }, ]) expect(snapshot.hasMore).toBe(false) - expect(snapshot.nodes.map(n => n.seq)).toEqual([1, 3, 7, 9]) + expect(eventSeqs(session)).toEqual([...older, ...newer].map(event => event.seq)) }) it('installs a page without interpreting business replacement metadata', async () => { @@ -416,7 +195,7 @@ describe('paging', () => { await session.open() const snapshot = session.getSnapshot() expect(snapshot.openState).toBe('open') - expect(chatSeqs(snapshot)).toEqual([80, 81, 82]) + expect(eventSeqs(session)).toEqual([80, 81, 82]) expect(errorSpy).not.toHaveBeenCalled() } finally { errorSpy.mockRestore() @@ -431,10 +210,10 @@ describe('paging', () => { const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined) try { await session.open() - const nodesBefore = session.getSnapshot().nodes + const windowBefore = session.eventSource.getSnapshot() await session.loadOlder() const snapshot = session.getSnapshot() - expect(snapshot.nodes).toEqual(nodesBefore) + expect(session.eventSource.getSnapshot().entries).toEqual(windowBefore.entries) expect(snapshot.hasMore).toBe(false) } finally { errorSpy.mockRestore() @@ -532,40 +311,41 @@ describe('prompt and cancel errors', () => { expect(api.callsOf('session.cancel')).toEqual([]) }) - it('sends content through session.prompt; composerPhase steps blank → engaging synchronously at send entry', async () => { + it('publishes the first-prompt lifecycle synchronously before the Remote settles', async () => { const { api, session } = makeSession() session.handleBlank(true) - // The blank → engaging edge fires before the RPC settles: the first-send - // flow reads the phase on the session area's first frame to keep the - // guidance hero from flashing back in. - expect(session.getSnapshot().composerPhase).toBe('blank') + expect(session.getSnapshot()).toMatchObject({ + blank: true, promptAttempted: false, awaitingFirstTurn: false, + }) const inFlight = session.prompt([{ type: 'text', text: '要发的' }], 'queue') - expect(session.getSnapshot().composerPhase).toBe('engaging') + expect(session.getSnapshot()).toMatchObject({ + blank: true, promptAttempted: true, awaitingFirstTurn: true, + }) const result = await inFlight expect(result.ok).toBe(true) - // Monotone: settlement alone does not step the phase anywhere. - expect(session.getSnapshot().composerPhase).toBe('engaging') + expect(session.getSnapshot()).toMatchObject({ + blank: false, promptAttempted: true, awaitingFirstTurn: true, + }) expect(api.callsOf('session.prompt')).toMatchObject([{ sessionId: SID, mode: 'queue', content: [{ type: 'text', text: '要发的' }], clientTimeZone: new Intl.DateTimeFormat().resolvedOptions().timeZone, }]) - // First content lands (running turn): engaging → active. session.handleRunning(true) - expect(session.getSnapshot().composerPhase).toBe('active') + expect(session.getSnapshot()).toMatchObject({ running: true, awaitingFirstTurn: false }) }) - it('business failure lands in promptError with op=send; the phase stays engaging (retry, no hero bounce)', async () => { + it('keeps the attempted-first-prompt state when the Host rejects the prompt', async () => { const { api, session } = makeSession() session.handleBlank(true) api.onPrompt = () => Promise.resolve(err({ code: 'agent-busy', message: 'busy', details: { reason: 'x' } })) const result = await session.prompt([{ type: 'text', text: '失败的' }], 'queue') expect(result.ok).toBe(false) expect(session.getSnapshot().promptError).toMatchObject({ op: 'send', error: { code: 'agent-busy' } }) - // Failed first prompt: composer + error strip is the retry surface — - // blank is unreachable once a send was initiated. - expect(session.getSnapshot().composerPhase).toBe('engaging') + expect(session.getSnapshot()).toMatchObject({ + blank: true, promptAttempted: true, awaitingFirstTurn: true, + }) }) it('lands cancel failures in promptError with op=stop', async () => { @@ -645,7 +425,7 @@ describe('remaining branches', () => { // err result: window unchanged api.onHistory = () => Promise.resolve(err({ code: 'internal', message: 'x', details: {} })) await session.loadOlder() - expect(session.getSnapshot().nodes).toHaveLength(2) + expect(eventSeqs(session)).toHaveLength(6) expect(session.getSnapshot().hasMore).toBe(true) // empty page: hasMore adopts the response api.onHistory = () => histResponse([], false) @@ -700,7 +480,7 @@ describe('remaining branches', () => { expect(snapshot.openError).toMatchObject({ code: 'internal', message: 'session event stream page did not end at its requested cursor', }) - expect(snapshot.nodes).toEqual([]) + expect(eventSeqs(session)).toEqual([]) }) it('deduplicates repeated running flips and records removal', () => { @@ -715,11 +495,11 @@ describe('remaining branches', () => { it('drops live events while cold/error (no window upkeep)', async () => { const { api, session } = makeSession() await follow(api, ev.user(0, '冷态帧')) - expect(session.getSnapshot().nodes).toEqual([]) + expect(eventSeqs(session)).toEqual([]) api.onHistory = () => Promise.resolve(err({ code: 'internal', message: 'x', details: {} })) await session.open() await follow(api, ev.user(0, '错态帧')) - expect(session.getSnapshot().nodes).toEqual([]) + expect(eventSeqs(session)).toEqual([]) }) it('preserves a Host-reported failure that terminates the live source', async () => { @@ -757,7 +537,7 @@ describe('remaining branches', () => { await deliveries await vi.waitFor(() => { expect(session.getSnapshot().openState).toBe('error') }) expect(session.getSnapshot().openError).toMatchObject({ code: 'internal', message: 'repair wire down' }) - expect(session.getSnapshot().nodes).toHaveLength(2) + expect(eventSeqs(session)).toHaveLength(6) }) it('doOpen transport throw of a stale generation is swallowed (generation guard in catch)', async () => { @@ -785,7 +565,7 @@ describe('remaining branches', () => { modelSelection: { provider: 'deepseek-official', model: 'stale' }, })) // success, but its generation is gone await Promise.all([opening, resynced]) - expect(session.getSnapshot().nodes.map(n => n.seq)).toEqual([7, 9]) // only the fresh generation's window + expect(eventSeqs(session)).toEqual(plainTurn(6, 1, '新', '代').map(event => event.seq)) }) it('drops a gap repair superseded by a full resync while its pull was in flight', async () => { @@ -804,7 +584,7 @@ describe('remaining branches', () => { modelSelection: { provider: 'deepseek-official', model: 'stale' }, })) // repair result: stale, dropped await Promise.all([delivery, resynced]) - expect(session.getSnapshot().nodes.map(n => n.seq)).toEqual([7, 9]) + expect(eventSeqs(session)).toEqual(plainTurn(6, 1, 'c', 'd').map(event => event.seq)) }) it('successful cancel leaves no promptError', async () => { @@ -821,7 +601,7 @@ describe('remaining branches', () => { await expect(session.dispose()).resolves.toBeUndefined() }) - it('carries history-entry and follow-frame views into the business-neutral Event input', async () => { + it('carries history-entry and follow-frame views through the event feed', async () => { const { api, session } = makeSession() const callView = { for: 'call', view: { card: 'generic', title: '历史卡' } } api.onHistory = () => Promise.resolve(ok({ @@ -834,7 +614,7 @@ describe('remaining branches', () => { modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' }, })) await session.open() - expect(chatEvents(session.getSnapshot()).slice(-2).map(item => item.view)).toEqual([ + expect(windowEntries(session).slice(-2).map(item => item.view)).toEqual([ callView, { for: 'result', view: { card: 'generic', title: '历史果' } }, ]) @@ -843,7 +623,7 @@ describe('remaining branches', () => { ev.toolCall(8, 2, 'l1', 'write', '{}'), { for: 'call', view: { card: 'generic', title: '直播卡' } }, ) - expect(chatEvents(session.getSnapshot()).at(-1)?.view).toEqual({ + expect(windowEntries(session).at(-1)?.view).toEqual({ for: 'call', view: { card: 'generic', title: '直播卡' }, }) await follow( @@ -851,22 +631,63 @@ describe('remaining branches', () => { ev.toolResult(9, 2, 'l1', 'ok'), { for: 'result', view: { card: 'generic', title: '直播果' } }, ) - expect(chatEvents(session.getSnapshot()).at(-1)?.view).toEqual({ + expect(windowEntries(session).at(-1)?.view).toEqual({ for: 'result', view: { card: 'generic', title: '直播果' }, }) }) }) describe('resync', () => { - it('rebuilds the window; cold instances no-op', async () => { + it('keeps the old feed until one sorted page-and-live replacement is ready', async () => { + const { api, session } = makeSession() + api.onHistory = () => histResponse(plainTurn(0, 0, '旧', '窗')) + await session.open() + const oldWindow = session.eventSource.getSnapshot() + const replacement = deferred>>() + api.followCursor = 15 + api.onHistory = () => replacement.promise + const publications: ReturnType[] = [] + const off = session.eventSource.subscribe(() => { + publications.push(session.eventSource.getSnapshot()) + }) + + const syncing = session.resync() + await vi.waitFor(() => { expect(api.callsOf('session.history')).toHaveLength(2) }) + expect(session.eventSource.getSnapshot()).toBe(oldWindow) + expect(publications).toEqual([]) + + await Promise.all([ + follow(api, ev.user(17, '后到高位')), + follow(api, ev.user(16, '后到低位')), + ]) + expect(session.eventSource.getSnapshot()).toBe(oldWindow) + replacement.resolve(ok({ + events: entries(plainTurn(10, 2, '终', '页')) as never[], + hasMore: false, + modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' }, + })) + await syncing + + expect(publications).toHaveLength(1) + expect(publications[0]?.entries).not.toHaveLength(0) + expect(publications[0]?.change.kind).toBe('replace') + expect(eventSeqs(session)).toEqual([10, 11, 12, 13, 14, 15, 16, 17]) + off() + }) + + it('rebuilds the window without clearing control state; cold instances no-op', async () => { const { api, session } = makeSession() api.onHistory = () => histResponse(plainTurn(0, 0, 'a', 'b')) await session.open() + session.handleRunning(true) + session.handleAgentError('still visible') api.onHistory = () => histResponse([...plainTurn(0, 0, 'a', 'b'), ...plainTurn(6, 1, 'c', 'd')]) await session.resync() const snapshot = session.getSnapshot() expect(snapshot.openState).toBe('open') - expect(snapshot.nodes).toHaveLength(4) + expect(snapshot.running).toBe(true) + expect(snapshot.lastAgentError).toBe('still visible') + expect(eventSeqs(session)).toHaveLength(12) const cold = makeSession() await cold.session.resync() @@ -885,55 +706,24 @@ describe('resync', () => { await resynced const snapshot = session.getSnapshot() expect(snapshot.openState).toBe('open') // stale failure did not settle the fresh generation into error - expect(snapshot.nodes.map(n => n.seq)).toEqual([7, 9]) + expect(eventSeqs(session)).toEqual(plainTurn(6, 1, '新', '代').map(event => event.seq)) }) }) -describe('reference stability (the memo contract)', () => { - it('keeps unchanged node references across an append and swaps the snapshot object', async () => { +describe('snapshot ownership', () => { + it('publishes event-window appends without changing an unrelated Session snapshot', async () => { const { api, session } = makeSession() api.onHistory = () => histResponse(plainTurn(0, 0, '稳', '定')) await session.open() - const before = session.getSnapshot() - const firstKey = before.chat.order[0]! - const secondKey = before.chat.order[1]! - const first = before.chat.nodes.get(firstKey) - const second = before.chat.nodes.get(secondKey) + const sessionBefore = session.getSnapshot() + const windowBefore = session.eventSource.getSnapshot() + const firstEntry = windowBefore.entries[0] await follow(api, ev.user(6, '追加')) - const after = session.getSnapshot() - expect(after).not.toBe(before) // top-level swap on change - expect(after.chat.nodes.get(firstKey)).toBe(first) - expect(after.chat.nodes.get(secondKey)).toBe(second) - expect(after.chat.order).toHaveLength(7) - // No change → same snapshot reference. - expect(session.getSnapshot()).toBe(after) - }) - - it('keeps unrelated Session arrays and settled Chat Nodes stable across Event updates', async () => { - const { api, session } = makeSession() - api.onHistory = () => histResponse(plainTurn(0, 0, '底', '座')) - await session.open() - await Promise.all([ - follow(api, ev.turnStart(6, 1)), - follow(api, ev.stepStart(7, 1)), - follow(api, ev.toolCall(8, 1, 'c1', 'echo', '{}')), - ]) - const before = session.getSnapshot() - const settledKey = before.chat.order[0]! - const settledNode = before.chat.nodes.get(settledKey) - await Promise.all([ - follow(api, ev.chunkStart(9, 1)), - follow(api, ev.chunkText(10, 1, '与工具无关的流式')), - ]) - const after = session.getSnapshot() - expect(after).not.toBe(before) - expect(after.runningCalls).toBe(before.runningCalls) - expect(after.chat.nodes.get(settledKey)).toBe(settledNode) - await follow(api, ev.toolResult(11, 1, 'c1', 'ECHO')) - const resolved = session.getSnapshot() - expect(resolved.chat.nodes.get(settledKey)).toBe(settledNode) - await follow(api, ev.assistant(12, 1, '完成')) - expect(session.getSnapshot()).not.toBe(resolved) + const windowAfter = session.eventSource.getSnapshot() + expect(session.getSnapshot()).toBe(sessionBefore) + expect(windowAfter).not.toBe(windowBefore) + expect(windowAfter.entries[0]).toBe(firstEntry) + expect(windowAfter.change).toMatchObject({ kind: 'append' }) }) }) diff --git a/packages/client/runtime/tests/sessions-service.client.spec.ts b/packages/api/session-controller/tests/sessions-service.client.spec.ts similarity index 81% rename from packages/client/runtime/tests/sessions-service.client.spec.ts rename to packages/api/session-controller/tests/sessions-service.client.spec.ts index 7285b25b0e..461190f3ce 100644 --- a/packages/client/runtime/tests/sessions-service.client.spec.ts +++ b/packages/api/session-controller/tests/sessions-service.client.spec.ts @@ -1,7 +1,7 @@ /** - * SessionRuntime: list store projection (manager → {ids, byId, current} - * with derived titles), the migrated current-selection account (open - * validation, persisted mask semantics, cell resolution), scope-tree + * ClientSessions: list store projection (manager → {ids, byId, current} + * with derived titles), the current-selection account (open validation and + * persisted mask semantics), scope-tree * lifecycle (lazy mint / frozen survival / removed teardown with staged * deferral — the stage follows list.current), binding identity, breadcrumb * projection, create. @@ -9,21 +9,31 @@ import { Context } from '@deepseek-ai/cordis' import { afterEach, describe, expect, it, vi } from 'vitest' import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client' -import { SessionCreateError, SessionRuntime, scopeOf } from '../src/client/sessions/service.ts' -import { FakeApiClient, deferred, err, fakeRemote, ok } from './fake-api.client.ts' +import { ClientSessions, SessionCreateError } from '../src/client/sessions/service.ts' +import { scopeOf } from '../src/client/scope.ts' +import type { SessionFollowFrame } from '../src/types.ts' +import { + FakeApiClient, + deferred, + err, + fakeRemote, + ok, + type RuntimeRemotes, +} from './fake-api.client.ts' const sid = (s: string): SessionId => s as SessionId interface Bench { ctx: Context api: FakeApiClient - svc: SessionRuntime + svc: ClientSessions } -function bench(): Bench { +function bench(configureRemote?: (remote: RuntimeRemotes) => RuntimeRemotes): Bench { const ctx = new Context() const api = new FakeApiClient() - const svc = new SessionRuntime(ctx, api, fakeRemote(api)) + const remote = fakeRemote(api) + const svc = new ClientSessions(ctx, api, configureRemote?.(remote) ?? remote) return { ctx, api, svc } } @@ -147,7 +157,7 @@ describe('scope tree', () => { expect(scopeOf(b.ctx)).toBeUndefined() const binding = b.svc.binding(sid('s1')) b.svc.open(sid('s1')) - expect(binding?.session).toBe(b.svc.currentProvideInfo.getSnapshot().hooks['session']) + expect(b.svc.sessionOf(scoped as Context)).toBe(binding?.session) expect(b.svc.binding(sid('s1'))).toBe(binding) expect(binding?.ctx).toBe(scoped) }) @@ -214,6 +224,166 @@ describe('scope tree', () => { }) }) +describe('Agent scope disposal lifecycle', () => { + it('root disposal runs Agent scope effects', async () => { + const b = bench() + const readiness = b.ctx.plugin(() => undefined) + await readiness + b.svc.handleSessionAdded({ + sessionId: sid('live'), updatedAt: 1, running: false, blank: true, + }) + await Promise.resolve() + const scoped = b.svc.scope(sid('live')) + if (scoped === undefined) throw new Error('fixture Agent Context was not minted') + await scoped.fiber.await() + const scopeDisposed = vi.fn() + scoped.effect(() => scopeDisposed, 'fixture Agent scope effect') + await b.ctx.fiber.dispose() + + expect(scopeDisposed).toHaveBeenCalledOnce() + expect(b.svc.sessionOf(scoped)).toBeUndefined() + }) + + it('root disposal waits for an opened Session source to finish closing', async () => { + const closeGate = deferred() + const abortObserved = vi.fn() + let followSignal: AbortSignal | undefined + const b = bench(remote => ({ + ...remote, + session: { + ...remote.session, + follow: (_request, signal) => { + if (signal === undefined) throw new Error('fixture requires a signal') + followSignal = signal + let opened = false + return { + [Symbol.asyncIterator]: () => ({ + next: () => { + if (!opened) { + opened = true + return Promise.resolve({ + done: false, + value: { type: 'opened', cursor: -1 } as const, + }) + } + return new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { + abortObserved() + void closeGate.promise.then(() => { + reject(signal.reason instanceof Error + ? signal.reason + : new Error(String(signal.reason))) + }) + }, { once: true }) + }) + }, + }), + } + }, + }, + })) + const readiness = b.ctx.plugin(() => undefined) + await readiness + await feedList(b, [{ id: 's1' }]) + b.svc.open(sid('s1')) + await vi.waitFor(() => { + expect(b.svc.binding(sid('s1'))?.session.getSnapshot().openState).toBe('open') + }) + + const disposal = b.ctx.fiber.dispose() + const settled = vi.fn() + const observed = disposal.then(settled) + + await vi.waitFor(() => { expect(abortObserved).toHaveBeenCalledOnce() }) + expect(followSignal?.aborted).toBe(true) + expect(settled).not.toHaveBeenCalled() + + closeGate.resolve(undefined) + await observed + expect(settled).toHaveBeenCalledOnce() + }) + + it('root disposal joins every Session drop already started by pruning under load', async () => { + const closeGates = new Map>>() + const aborted = new Set() + const b = bench(remote => ({ + ...remote, + session: { + ...remote.session, + follow: (request, signal) => { + if (signal === undefined) throw new Error('fixture requires a signal') + const sessionId = request.address.kind === 'session' + ? request.address.sessionId + : request.address.childSessionId + const closeGate = deferred() + closeGates.set(sessionId, closeGate) + let opened = false + return { + [Symbol.asyncIterator]: () => ({ + next: () => { + if (!opened) { + opened = true + return Promise.resolve({ + done: false, + value: { type: 'opened', cursor: -1 } as const, + }) + } + return new Promise>((_resolve, reject) => { + signal.addEventListener('abort', () => { + aborted.add(sessionId) + void closeGate.promise.then(() => { + reject(signal.reason instanceof Error + ? signal.reason + : new Error(String(signal.reason))) + }) + }, { once: true }) + }) + }, + }), + } + }, + }, + })) + const readiness = b.ctx.plugin(() => undefined) + await readiness + const sessionIds = Array.from({ length: 24 }, (_, index) => sid(`load-${String(index)}`)) + const retained = sessionIds.at(-1) + const held = sessionIds[0] + if (retained === undefined || held === undefined) throw new Error('fixture requires sessions') + await feedList(b, sessionIds.map(id => ({ id }))) + for (const id of sessionIds) b.svc.open(id) + await vi.waitFor(() => { + for (const id of sessionIds) { + expect(b.svc.binding(id)?.session.getSnapshot().openState).toBe('open') + } + }) + + const pruned = sessionIds.slice(0, -1) + await feedList(b, [{ id: retained }]) + await vi.waitFor(() => { expect(aborted.size).toBe(pruned.length) }) + for (const id of pruned) expect(b.svc.scope(id)).toBeUndefined() + + const disposal = b.ctx.fiber.dispose() + const settled = vi.fn() + const observed = disposal.then(settled) + await vi.waitFor(() => { expect(aborted.size).toBe(sessionIds.length) }) + + const otherClosures: Promise[] = [] + for (const [id, gate] of closeGates) { + if (id === held) continue + gate.resolve(undefined) + otherClosures.push(gate.promise) + } + await Promise.all(otherClosures) + await new Promise((resolve) => { setTimeout(resolve, 0) }) + expect(settled).not.toHaveBeenCalled() + + closeGates.get(held)?.resolve(undefined) + await observed + expect(settled).toHaveBeenCalledOnce() + }) +}) + describe('current selection (migrated from ui-layout, arbitrated into the list snapshot)', () => { afterEach(() => { vi.unstubAllGlobals() }) @@ -274,78 +444,7 @@ describe('current selection (migrated from ui-layout, arbitrated into the list s }) }) -describe('cell (render-layer session kit)', () => { - it('resolves an identity-stable {sessionId, session} cell through the current projection', async () => { - const b = bench() - await feedList(b, [{ id: 's1' }]) - b.svc.open(sid('s1')) - const info = b.svc.currentProvideInfo.getSnapshot() - expect(info.sessionId).toBe('s1') - // The bundle carries bare observables; hook binding happens in React. - expect(info.hooks['session']).toBe(b.svc.binding(sid('s1'))?.session) - // Re-staging the same id republishes nothing: identity holds. - b.svc.open(sid('s1')) - expect(b.svc.currentProvideInfo.getSnapshot()).toBe(info) - }) - - it('currentProvideInfo follows selection: absent projection ↔ definite bundle, notified on each move', async () => { - const b = bench() - await feedList(b, [{ id: 's1' }, { id: 's2' }]) - const absent = b.svc.currentProvideInfo.getSnapshot() - expect(absent.sessionId).toBeUndefined() - expect(Object.hasOwn(absent.hooks, 'session')).toBe(true) - const notified = vi.fn() - b.svc.currentProvideInfo.subscribe(notified) - b.svc.open(sid('s1')) - const s1Bundle = b.svc.currentProvideInfo.getSnapshot() - expect(s1Bundle.sessionId).toBe('s1') - expect(s1Bundle.hooks['session']).toBe(b.svc.binding(sid('s1'))?.session) - expect(notified).toHaveBeenCalledTimes(1) - b.svc.open(sid('s2')) - const s2Bundle = b.svc.currentProvideInfo.getSnapshot() - expect(s2Bundle.sessionId).toBe('s2') - expect(s2Bundle).not.toBe(s1Bundle) - expect(notified).toHaveBeenCalledTimes(2) - b.svc.clear() - await Promise.resolve() // clearSelection projects through the manager notifier - expect(b.svc.currentProvideInfo.getSnapshot().sessionId).toBeUndefined() - }) - - it('a provider roster change under a stable current id republishes the bundle', async () => { - const b = bench() - await feedList(b, [{ id: 's1' }]) - b.svc.open(sid('s1')) - const before = b.svc.currentProvideInfo.getSnapshot() - const notified = vi.fn() - b.svc.currentProvideInfo.subscribe(notified) - const source = { getSnapshot: () => 'live', subscribe: () => () => {} } - const dispose = b.svc.provide({ - hooks: ['extra'], - props: ['marker'], - resolve: () => ({ hooks: { extra: source }, props: { marker: 7 } }), - }) - const added = b.svc.currentProvideInfo.getSnapshot() - expect(added).not.toBe(before) - expect(added).toMatchObject({ sessionId: 's1', props: { marker: 7 } }) - expect(added.hooks['extra']).toBe(source) - expect(notified).toHaveBeenCalledTimes(1) - dispose() - const removed = b.svc.currentProvideInfo.getSnapshot() - expect(removed).not.toBe(added) - expect(Object.hasOwn(removed.hooks, 'extra')).toBe(false) - expect(notified).toHaveBeenCalledTimes(2) - }) - - it('an unsubscribed currentProvideInfo listener stops receiving notifications', async () => { - const b = bench() - await feedList(b, [{ id: 's1' }]) - const notified = vi.fn() - const off = b.svc.currentProvideInfo.subscribe(notified) - off() - b.svc.open(sid('s1')) - expect(notified).not.toHaveBeenCalled() - }) - +describe('binding and stage lifecycle', () => { it('binding() is pure resolution: no staging, no deferred sweep', async () => { const b = bench() await feedList(b, [{ id: 's1' }, { id: 's2' }]) @@ -399,32 +498,6 @@ describe('cell (render-layer session kit)', () => { }) }) -describe('slot-store scope prune hook', () => { - it('notifies ctx.slots.pruneStoreScope when a scope dies (both teardown paths)', async () => { - const b = bench() - const pruneStoreScope = vi.fn() - b.ctx.reflect.provide('slots', { pruneStoreScope }) - await feedList(b, [{ id: 's1' }, { id: 's2' }]) - b.svc.scope(sid('s1')) - b.svc.scope(sid('s2')) - b.svc.open(sid('s2')) // s2 staged - await feedList(b, []) // s1 off stage → immediate drop; s2 staged → deferred - expect(pruneStoreScope).toHaveBeenCalledWith('s1') - expect(pruneStoreScope).not.toHaveBeenCalledWith('s2') - await feedList(b, [{ id: 's3' }]) - b.svc.open(sid('s3')) // stage moves → deferred sweep drops s2 - expect(pruneStoreScope).toHaveBeenCalledWith('s2') - }) - - it('tolerates a slots-less boot (object-layer benches carry no slot service)', async () => { - const b = bench() - await feedList(b, [{ id: 's1' }]) - b.svc.scope(sid('s1')) - await feedList(b, []) // teardown without ctx.slots must not throw - expect(b.svc.scope(sid('s1'))).toBeUndefined() - }) -}) - describe('catalog-addressed navigation', () => { it('uses catalog labels for a listed addressed route', async () => { const b = bench() diff --git a/packages/client/runtime/tests/subagent-lineage.client.spec.ts b/packages/api/session-controller/tests/subagent-lineage.client.spec.ts similarity index 91% rename from packages/client/runtime/tests/subagent-lineage.client.spec.ts rename to packages/api/session-controller/tests/subagent-lineage.client.spec.ts index 05881576bf..f9c8ec0d69 100644 --- a/packages/client/runtime/tests/subagent-lineage.client.spec.ts +++ b/packages/api/session-controller/tests/subagent-lineage.client.spec.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest' -import type { SessionId, SessionSummary } from '@deepseek-ai/dsh-client-runtime/client' -import { indexSubagentDescendants } from '@deepseek-ai/dsh-client-runtime/client' +import type { SessionId } from '@deepseek-ai/dsh-session/types' +import type { SessionSummary } from '../src/client/index.ts' +import { indexSubagentDescendants } from '../src/client/index.ts' const sid = (id: string) => id as SessionId diff --git a/packages/client/runtime/tests/time-zone.client.spec.ts b/packages/api/session-controller/tests/time-zone.client.spec.ts similarity index 92% rename from packages/client/runtime/tests/time-zone.client.spec.ts rename to packages/api/session-controller/tests/time-zone.client.spec.ts index d96c9476c1..983dabc20c 100644 --- a/packages/client/runtime/tests/time-zone.client.spec.ts +++ b/packages/api/session-controller/tests/time-zone.client.spec.ts @@ -5,7 +5,7 @@ afterEach(() => { vi.restoreAllMocks() }) -describe('browser time zone', () => { +describe('Session Controller browser time zone', () => { it('returns the runtime-resolved zone', () => { expect(resolvedClientTimeZone()).toBe( new Intl.DateTimeFormat().resolvedOptions().timeZone, diff --git a/packages/api/session-controller/tests/transport.client.spec.ts b/packages/api/session-controller/tests/transport.client.spec.ts index 2dd1c2602d..46c2064c9e 100644 --- a/packages/api/session-controller/tests/transport.client.spec.ts +++ b/packages/api/session-controller/tests/transport.client.spec.ts @@ -7,11 +7,10 @@ import { } from '@deepseek-ai/dsh-api-gateway/client' import type { RemoteResult } from '@deepseek-ai/dsh-typert-protocol' import { - apply, createSessionControlStream, SessionEventStream, sessionStreamFailure, - type SessionEventChange, + type SessionJournalChange, type SessionRemote, } from '../src/client/index.ts' import type { @@ -105,10 +104,6 @@ class ScriptedSessionRemote implements SessionTransportRemote { } describe('Session Client stream adapters', () => { - it('installs no Client service', () => { - apply() - }) - it('binds an event journal to one address and publishes replace, append, and prepend changes', async () => { const remote = new ScriptedSessionRemote( [{ @@ -124,7 +119,7 @@ describe('Session Client stream adapters', () => { { ok: true, value: page([entry(0), entry(1)], false) }, ], ) - const changes: SessionEventChange[] = [] + const changes: SessionJournalChange[] = [] const stream = new SessionEventStream(sessionClient(remote), ADDRESS, { publish: (change) => { changes.push(change) }, failed: vi.fn(), @@ -163,7 +158,7 @@ describe('Session Client stream adapters', () => { { ok: true, value: page([entry(0), entry(1), entry(2), entry(3), entry(4)]) }, ], ) - const changes: SessionEventChange[] = [] + const changes: SessionJournalChange[] = [] const carrierFailed = vi.fn() const stream = new SessionEventStream(sessionClient(remote), ADDRESS, { publish: (change) => { changes.push(change) }, diff --git a/packages/api/session-controller/tsconfig.client.json b/packages/api/session-controller/tsconfig.client.json index 20aedea93a..5703caf280 100644 --- a/packages/api/session-controller/tsconfig.client.json +++ b/packages/api/session-controller/tsconfig.client.json @@ -5,8 +5,8 @@ "outDir": "lib/types", "tsBuildInfoFile": "lib/tsconfig.client.tsbuildinfo" }, - "files": [ - "src/client/index.ts", + "include": [ + "src/client/**/*.ts", "src/types.ts", "src/remote-events.ts" ], @@ -14,12 +14,16 @@ { "path": "../../../vendor/cordis" }, { "path": "../gateway/tsconfig.client.json" }, { "path": "../../attachment/attachment" }, + { "path": "../../client/connection/tsconfig.client.json" }, + { "path": "../../client/store" }, { "path": "../../core/session" }, { "path": "../../jobs/jobs" }, { "path": "../../llm/llm" }, { "path": "../../session/session-projection" }, + { "path": "../../session/session-title" }, { "path": "../../core/tools" }, { "path": "../../util/brand" }, + { "path": "../../util/crypto" }, { "path": "../../workspace/workspace" }, { "path": "../../typert/protocol" } ] diff --git a/packages/api/session-controller/tsconfig.host.json b/packages/api/session-controller/tsconfig.host.json index 797854fbe0..efceb803df 100644 --- a/packages/api/session-controller/tsconfig.host.json +++ b/packages/api/session-controller/tsconfig.host.json @@ -26,6 +26,7 @@ { "path": "../../core/session" }, { "path": "../../core/tools" }, { "path": "../../attachment/attachment" }, + { "path": "../../interaction/permission-presets" }, { "path": "../../jobs/jobs" }, { "path": "../../llm/llm" }, { "path": "../../preset/agent-presets" }, diff --git a/packages/client/runtime/src/client/contract/sessions-port.ts b/packages/client/runtime/src/client/contract/sessions-port.ts deleted file mode 100644 index 1026d271b2..0000000000 --- a/packages/client/runtime/src/client/contract/sessions-port.ts +++ /dev/null @@ -1,47 +0,0 @@ -/** - * Cross-domain sessions face consumed by the workspace domain instead of the - * sessions implementation. The sessions domain satisfies it structurally — - * SessionRuntime is assignable, checked - * wherever the assembly layer or a test injects the real service — so - * widening this face is the explicit act of widening the inter-domain - * dependency. - */ - -import type { SessionId, WorkspaceId } from '@deepseek-ai/dsh-api-remotes/client' -import type { ObservableSnapshot } from './store.ts' - -/** Session-list row facts sibling domains read: recency, blank-reuse eligibility, and its cwd canon. */ -export interface SessionsPortSummary { - id: SessionId - /** Empty-log bit (blank sessions are reused by New Session instead of minting another). */ - blank: boolean - cwd?: string - updatedAt: number -} - -/** Session-list facts sibling domains read: readiness, selection, and the row map. */ -export interface SessionsPortList { - ids: SessionId[] - byId: Record - current: SessionId | undefined - phase: 'pending' | 'ready' -} - -/** The sessions-service face injected into sibling domains. */ -export interface SessionsPort { - /** Observable list snapshot (read face only; writes stay inside the sessions domain). */ - readonly list: ObservableSnapshot - /** - * Create a session on the host. - * @param opts - target workspace. - * @returns the new session id. - */ - create(opts: { workspaceId: WorkspaceId }): Promise - /** - * Select a session as current. - * @param id - session id (must exist in the list store). - */ - open(id: SessionId): void - /** Clear the current selection into the no-session view state. */ - clear(): void -} diff --git a/packages/client/runtime/src/client/sessions/remotes.ts b/packages/client/runtime/src/client/sessions/remotes.ts deleted file mode 100644 index 6479aee4be..0000000000 --- a/packages/client/runtime/src/client/sessions/remotes.ts +++ /dev/null @@ -1,12 +0,0 @@ -/** - * Remote namespaces the Session cluster calls. One parameter for one concept: - * the generated surface a Session and its manager reach the Host through. - * - * @module @deepseek-ai/dsh-client-runtime/client/sessions/remotes - */ - -import type { Context } from '@deepseek-ai/cordis' -import type {} from '@deepseek-ai/dsh-api-remotes/client' - -/** The generated Remote namespaces and Gateway stream factory a Session cluster uses. */ -export type SessionRemotes = Pick diff --git a/packages/client/runtime/tests/client-apply.client.spec.ts b/packages/client/runtime/tests/client-apply.client.spec.ts deleted file mode 100644 index 5846a3d454..0000000000 --- a/packages/client/runtime/tests/client-apply.client.spec.ts +++ /dev/null @@ -1,198 +0,0 @@ -/** - * Runtime plugin browser-half apply: slots + object services mounting over the - * connection handle, Remote stream wiring into the object layer, and - * fiber-scoped stream teardown. - */ -import { Context } from '@deepseek-ai/cordis' -import { describe, expect, it, vi } from 'vitest' -import type { ConnectionHandle } from '@deepseek-ai/dsh-api-remotes/client' -import { SESSION_SEARCH_RESULT_LIMIT } from '@deepseek-ai/dsh-api-session-controller/client' -import TypertRegistry from '@deepseek-ai/dsh-typert-registry' -import * as RuntimeClient from '../src/client/index.ts' -import type { ConversationNodeDefinition } from '../src/client/contract/conversation.ts' -import { scopeOf } from '../src/client/agents/scope.ts' -import { Session } from '../src/client/sessions/session.ts' -import { SessionRuntime } from '../src/client/sessions/service.ts' -import { FakeApiClient, fakeRemote, ok } from './fake-api.client.ts' - -interface Bench { - ctx: Context - api: FakeApiClient - runtime: { dispose(): Promise } - start: ReturnType> - dispatchRemote(event: string, args: readonly unknown[]): void -} - -async function mount(configure?: (api: FakeApiClient) => void): Promise { - const ctx = new Context() - await ctx.plugin(TypertRegistry) - const api = new FakeApiClient() - configure?.(api) - const listeners = new Map void>>() - const dispatchRemote = (event: string, args: readonly unknown[]): void => { - for (const listener of listeners.get(event) ?? []) listener(...args as never[]) - } - const start = vi.fn(() => ({ stop: () => {} })) - const handle: ConnectionHandle = { - api, - isLoopback: true, - hostDescription: { - getSnapshot: () => undefined, - subscribe: () => () => {}, - }, - rpc: { - call: () => Promise.reject(new Error('unexpected generic RPC call')), - }, - registerGenerationSource: () => () => {}, - start, - } - const remote = fakeRemote(api) - ctx.reflect.provide('connection', handle) - ctx.reflect.provide('remote', { - ...remote, - $on: (event: string, listener: (...args: never[]) => void) => { - const eventListeners = listeners.get(event) ?? new Set() - eventListeners.add(listener) - listeners.set(event, eventListeners) - return () => { eventListeners.delete(listener) } - }, - }) - ctx.reflect.provide('remote.commands', remote.commands) - ctx.reflect.provide('remote.session', remote.session) - ctx.reflect.provide('remote.workspace', remote.workspace) - const runtime = await ctx.plugin(RuntimeClient).await() - return { ctx, api, runtime, start, dispatchRemote } -} - -async function flushMicrotasks(): Promise { - for (let i = 0; i < 12; i++) await Promise.resolve() -} - -describe('runtime client apply', () => { - it('materializes Host-addressed Agent scopes before the Session list arrives', async () => { - const bench = await mount() - const adapter = bench.ctx.typert.contexts.getClient('agent') - const first = adapter?.resolve('s-early') - - expect(first).toBeDefined() - expect(scopeOf(first as Context)).toBe('s-early') - expect(adapter?.resolve('s-early')).toBe(first) - }) - - it('refreshes Sessions on every Gateway connection generation', async () => { - const refresh = vi.spyOn(SessionRuntime.prototype, 'handleConnected') - const bench = await mount() - - bench.ctx.emit('connection/reset') - bench.ctx.emit('connection/reset') - - expect(refresh).toHaveBeenCalledTimes(2) - refresh.mockRestore() - }) - - it('mounts slots, Sessions, and Workspaces and routes their independent streams', async () => { - const bench = await mount() - expect(bench.ctx.get('slots') !== undefined).toBe(true) - // The built-in 'root' declaration ships with this package's SlotRegistry - // (the SlotMap 'root' merge lives here). - expect(bench.ctx.slots.spec('root')).toEqual({ kind: 'single', scope: 'root' }) - const sessions = bench.ctx.get('sessions') - const workspaces = bench.ctx.get('workspaces') - expect(sessions !== undefined).toBe(true) - expect(workspaces !== undefined).toBe(true) - // The bound the wire schema enforces, not a per-connection negotiation. - expect((sessions as SessionRuntime).searchResultLimit).toBe(SESSION_SEARCH_RESULT_LIMIT) - if (workspaces === undefined) throw new Error('WorkspaceRuntime missing after runtime apply') - expect(bench.start).not.toHaveBeenCalled() - - // Session Remote events reach the object layer and land in the list store. - bench.dispatchRemote('api-session/added', [{ - sessionId: 's-new', updatedAt: 1, running: false, blank: true, - }]) - await Promise.resolve() - expect((sessions as { list: { getSnapshot(): { ids: string[] } } }).list.getSnapshot().ids).toContain('s-new') - await flushMicrotasks() - bench.api.pushWorkspace({ - type: 'upsert', - workspace: { - workspaceId: 'w-new' as never, path: '/w/new', title: 'new', sessionIds: [], - createdAt: '2026-01-01T00:00:00.000Z', updatedAt: '2026-01-01T00:00:00.000Z', - }, - }) - await flushMicrotasks() - expect(workspaces.list.getSnapshot().items[0]?.workspaceId).toBe('w-new') - // Gateway generation publication routes without throwing. - bench.ctx.emit('connection/reset') - }) - - it('selects the recent Workspace once when the first baselines have no current session', async () => { - const bench = await mount((api) => { - api.workspaceBaseline = { - items: [{ - workspaceId: 'w-recent', path: '/w/recent', title: 'recent', sessionIds: [], - createdAt: '2026-01-01T00:00:00.000Z', updatedAt: '2026-01-01T00:00:00.000Z', - }] as never[], - archivedSessionIds: [], - } - api.onList = () => Promise.resolve(ok({ items: [] })) - }) - - bench.ctx.emit('connection/reset') - - const sessions = bench.ctx.get('sessions') as SessionRuntime - await vi.waitFor(() => { - expect(bench.api.callsOf('session.create')).toEqual([{ workspaceId: 'w-recent' }]) - }) - expect(sessions.list.getSnapshot().current).toBe('fk-new') - - sessions.clear() - bench.api.pushWorkspace({ - type: 'upsert', - workspace: bench.api.workspaceBaseline.items[0] as never, - }) - await flushMicrotasks() - expect(sessions.list.getSnapshot().current).toBeUndefined() - expect(bench.api.callsOf('session.create')).toHaveLength(1) - }) - - it('wires registry changes into resident Sessions during the runtime apply pass', async () => { - const bench = await mount() - const sessions = bench.ctx.get('sessions') as SessionRuntime - bench.dispatchRemote('api-session/added', [{ - sessionId: 's-registry', updatedAt: 1, running: false, blank: true, - }]) - await flushMicrotasks() - expect(sessions.binding('s-registry' as never)).toBeDefined() - const rebuild = vi.spyOn(Session.prototype, 'rebuildConversationRegistry') - const definition: ConversationNodeDefinition = { - kind: 'registry-probe', - target: 'chat', - match: () => null, - start: () => null, - update: context => context.state, - buildViewNode: () => null, - } - - bench.ctx.conversationEvents.register(definition) - await flushMicrotasks() - - expect(rebuild).toHaveBeenCalledOnce() - rebuild.mockRestore() - }) - - it('does not own the Connection loop and closes its Remote streams on unload', async () => { - const bench = await mount() - const sessions = bench.ctx.get('sessions') as SessionRuntime - bench.dispatchRemote('api-session/added', [{ - sessionId: 's-open', updatedAt: 1, running: false, blank: false, - }]) - await flushMicrotasks() - sessions.open('s-open' as never) - await vi.waitFor(() => { expect(bench.api.activeFollows('s-open' as never)).toBe(1) }) - - await bench.runtime.dispose() - - expect(bench.start).not.toHaveBeenCalled() - expect(bench.api.activeFollows('s-open' as never)).toBe(0) - }) -})