mengzhx/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness.git synced 2026-08-29 04:26:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(connection): own RPC transport contracts

Browse Source
This commit is contained in:
imccyu
2026-08-27 22:26:29 +08:00
parent 3b40a14555
commit e14d354e83
52 changed files with 645 additions and 799 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/cli/tests/web-auth.e2e.ts
+9 -9
View File
@@ -119,19 +119,19 @@ async function stopWeb(running: RunningWeb): Promise<void> {
clearTimeout(forced)
}
/** POST one real API Proxy envelope while controlling the wire Host header. */
function describeHost(port: number, host: string, cookie?: string): Promise<HttpResult> {
/** POST one real Remote envelope while controlling the wire Host header. */
function describeSettings(port: number, host: string, cookie?: string): Promise<HttpResult> {
const body = JSON.stringify({
type: 'client-request',
rpcId: 'web-auth-real-cli',
method: 'host.describe',
payload: {},
method: 'settings/describe',
payload: { args: {} },
})
return new Promise((resolve, reject) => {
const req = httpRequest({
hostname: '127.0.0.1',
port,
path: '/api/host.describe',
path: '/api/settings/describe',
method: 'POST',
headers: {
host,
@@ -165,7 +165,7 @@ describe('dsh web authentication through the real CLI', () => {
expect(firstUrl.pathname).toBe('/')
expect(firstUrl.searchParams.get('token')).toMatch(/^[A-Za-z0-9_-]{43}$/u)
expect(await describeHost(port, `localhost:${String(port)}`)).toEqual({
expect(await describeSettings(port, `localhost:${String(port)}`)).toEqual({
status: 401,
body: 'unauthorized',
})
@@ -180,13 +180,13 @@ describe('dsh web authentication through the real CLI', () => {
expect(setCookie).not.toContain('Secure')
const cookie = setCookie.split(';', 1)[0]!
const authenticated = await describeHost(port, firstUrl.host, cookie)
const authenticated = await describeSettings(port, firstUrl.host, cookie)
expect(authenticated.status).toBe(200)
const authenticatedBody = JSON.parse(authenticated.body) as unknown
expect(authenticatedBody).toMatchObject({
type: 'server-response',
rpcId: 'web-auth-real-cli',
result: { ok: true, value: { version: expect.any(String) as unknown } },
result: { ok: true, value: { namespaces: expect.any(Array) as unknown } },
})
await stopWeb(first)
@@ -194,7 +194,7 @@ describe('dsh web authentication through the real CLI', () => {
second = await startWeb(root, dshHome, port)
const secondUrl = new URL(second.launchUrl)
expect(secondUrl.searchParams.get('token')).not.toBe(firstUrl.searchParams.get('token'))
expect((await describeHost(port, secondUrl.host, cookie)).status).toBe(200)
expect((await describeSettings(port, secondUrl.host, cookie)).status).toBe(200)
const credentialMode = (await stat(join(dshHome, '.credentials.yaml'))).mode & 0o777
expect(credentialMode).toBe(0o600)
apps/web/tests/assembled-boot.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// Shared scaffolding for the assembled-jsdom snapshots: the real built
// workspace `lib/client.js` artifacts booted through AppWebEntry's
// ModuleLoader path (loadBundle) against the keyless FixtureApiClient
// ModuleLoader path (loadBundle) against the keyless fixture Connection RPC
// transport. Every file that mounts this graph needs the same boot entry list,
// the same bundle map, the same jsdom globals, and the same mount call, and
// differs only in what it asserts afterwards, so the scaffolding lives here.
apps/web/tests/built-boot.expected.e2e.ts
+1 -1
View File
@@ -4,7 +4,7 @@
// reach a surface only the built bundles expose; this one asserts that the
// graph assembles at all — staged activation across the immediately tier and
// the inject layers, per-plugin CSS injection, and a rendered journey reaching
// chat content from the keyless FixtureApiClient transport.
// chat content from the keyless fixture Connection RPC.
//
// Component behavior remains owned by per-package suites (SlotTestRuntime
// benches over src). This smoke additionally pins the resident interaction
apps/web/tests/command-image-envelope.expected.e2e.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @vitest-environment jsdom
// The command image-attachment envelope over the BUILT client graph (real
// bundles via AppWebEntry, keyless FixtureApiClient transport): an enter
// bundles via AppWebEntry, keyless fixture Connection RPC): an enter
// submission carrying composer images resolves only through a command whose
// descriptor declares `input.images`. A non-declaring command refuses with
// one composer error banner and everything retained; a declaring command
apps/web/tests/goal-bar.e2e.ts
+1 -1
View File
@@ -1,5 +1,5 @@
// Keyless assembled-browser coverage for the goal bar over the shipped Web
// bundles and FixtureApiClient wire. The command creates a real projected
// bundles and the fixture Connection RPC. The command creates a real projected
// goal in the fixture session; the golden pins the active strip, while the
// clear gesture proves the acknowledged tombstone leaves neither stale chrome
// nor a duplicate-mutation error.
apps/web/tests/image-display.expected.e2e.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @vitest-environment jsdom
// Multimodal image surfaces over the BUILT client graph (the code-mode-fixture
// idiom: real bundles via AppWebEntry, keyless FixtureApiClient transport).
// idiom: real bundles via AppWebEntry, keyless fixture Connection RPC).
// Opens the fixture history session whose turn 73 carries an image in BOTH a
// user message and an assistant message, and pins the product surfaces: the
// history ImageGallery loading real fixture bytes through the authorized
apps/web/tests/max-tokens-notice.expected.e2e.ts
+1 -1
View File
@@ -1,7 +1,7 @@
// @vitest-environment jsdom
// Assembled max-tokens snapshot: boots the real built `packages/client/*/lib/
// client.js` bundles through AppWebEntry's ModuleLoader path against the
// keyless FixtureApiClient transport, opens the fixture session, and pins the
// keyless fixture Connection RPC, opens the fixture session, and pins the
// surface its max-tokens turn (72) reaches — the turn-end notice row that a
// provider output-cap truncation must render instead of ending silently.
//
apps/web/tests/replay-round-trip.e2e.ts
+1 -1
View File
@@ -200,7 +200,7 @@ describe('web e2e: fresh round trip through the real assembly', () => {
it.skipIf(MODE === 'record')('expands and collapses the reasoning fold from its click target', async () => {
onTestFailed(() => saveFailureShot(page, 'web-e2e-round-trip-think'))
// Interaction over the REAL wire-delivered transcript (the fixture-client
// tier pins the same gesture against FixtureApiClient; this one runs on
// tier pins the same gesture against the fixture Connection RPC; this one runs on
// follow-stream-fed state). Runs after the golden capture so the committed
// aria surface stays the untouched settled state.
await expandTurnProcesses(page)
apps/web/tests/search-card.expected.e2e.ts
+1 -1
View File
@@ -1,7 +1,7 @@
// @vitest-environment jsdom
// Assembled search-card snapshot: boots the real built workspace client bundles
// through AppWebEntry's ModuleLoader path against the keyless
// FixtureApiClient transport (no API key, no model round), opens the fixture
// fixture Connection RPC (no API key, no model round), opens the fixture
// session, and pins the search card the `grep` turn (fixture turn 67) renders in
// the assembled application. The built-boot smoke proves the graph boots but
// intentionally carries no behavior assertions; this is the assembled-output check
apps/web/tests/submission-echo.e2e.ts
+1 -1
View File
@@ -1,5 +1,5 @@
// @vitest-environment jsdom
// Local submission echo over the BUILT client graph (keyless FixtureApiClient
// Local submission echo over the BUILT client graph (keyless fixture Connection RPC
// transport): a text-plus-image send paints its echo bubble synchronously on
// the submit keystroke — before serialization, transport, or the fixture's
// durable admission — with the composer already cleared and editable, and the
apps/web/tests/todo-row.expected.e2e.ts
+1 -1
View File
@@ -1,7 +1,7 @@
// @vitest-environment jsdom
// Assembled todo snapshot: boots the real built `packages/client/*/lib/
// client.js` bundles through AppWebEntry's ModuleLoader path against the
// keyless FixtureApiClient transport, opens the fixture session, and pins the
// keyless fixture Connection RPC, opens the fixture session, and pins the
// two surfaces the fixture's parallel plan (turn 74, two items `in_progress`)
// reaches — the `todo_write` tool row and the dock's plan strip.
//
apps/web/tests/trajectory-image-display.expected.e2e.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @vitest-environment jsdom
// Trajectory image surfaces over the BUILT client graph (the code-mode-fixture
// idiom: real bundles via AppWebEntry, keyless FixtureApiClient transport).
// idiom: real bundles via AppWebEntry, keyless fixture Connection RPC).
// Opens the fixture history session whose turn 73 carries an image in BOTH a
// user message and an assistant message, and pins the Trajectory surfaces:
// selecting the ledger record renders the shared ui-attachment gallery from
packages/api/gateway/tests/gateway.client.spec.ts
+1 -2
View File
@@ -1809,7 +1809,7 @@ describe('Client Typert API', () => {
})
})
it('publishes the Fixture Host description after Remote events report ready', async () => {
it('publishes the Fixture Host facts after Remote events report ready', async () => {
const locationDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'location')
Object.defineProperty(globalThis, 'location', {
configurable: true,
@@ -1824,7 +1824,6 @@ describe('Client Typert API', () => {
if (connection === undefined) throw new Error('fixture Connection service is unavailable')
await vi.waitFor(() => {
expect(connection.hostDescription.getSnapshot()?.home).toBe('/home/fixture')
expect(connection.generation.getSnapshot()?.host.home).toBe('/home/fixture')
})
} finally {
packages/api/remotes/src/client/index.ts
+2 -2
View File
@@ -54,8 +54,8 @@ export type {} from '@deepseek-ai/dsh-api-session-controller/types'
* the carrier's runtime values stay behind their own module edge.
*/
export type {
ConnectionHandle, ConnectionSinks, ContentBlock, IApiClient,
MessageId, ModelCatalogFailure, ModelProviderGroup, ModelReasoningEffort, ModelSelection,
ConnectionHandle, ConnectionSinks, ContentBlock,
MessageId,
RpcError, RpcId, RpcRequest, RpcResponse, RpcResult, SessionId,
StreamChunk,
} from '@deepseek-ai/dsh-client-connection/client'
packages/client/connection/package.json
+5 -4
View File
@@ -1,6 +1,6 @@
{
"name": "@deepseek-ai/dsh-client-connection",
"description": "Wire consumer layer: HTTP client, generation lifecycle, and fixture API",
"description": "Authenticated RPC transport, generation lifecycle, and browser fixture",
"version": "0.1.1-rc.2",
"publishConfig": {
"access": "public"
@@ -38,7 +38,8 @@
},
"license": "MIT",
"dependencies": {
"@deepseek-ai/schemastery": "workspace:^"
"@deepseek-ai/schemastery": "workspace:^",
"zod": "^4.4.3"
},
"files": [
"lib/index.js",
@@ -51,7 +52,7 @@
"@deepseek-ai/dsh-attachment": "workspace:^",
"@deepseek-ai/dsh-commands": "workspace:^",
"@deepseek-ai/dsh-credentials": "workspace:^",
"@deepseek-ai/dsh-host-apiproxy": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-host-directory-picker": "workspace:^",
"@deepseek-ai/dsh-host-webserver": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
@@ -65,7 +66,7 @@
"@deepseek-ai/dsh-attachment": "workspace:^",
"@deepseek-ai/dsh-commands": "workspace:^",
"@deepseek-ai/dsh-credentials": "workspace:^",
"@deepseek-ai/dsh-host-apiproxy": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-host-directory-picker": "workspace:^",
"@deepseek-ai/dsh-host-webserver": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
packages/client/connection/src/client/api.ts
+15 -32
View File
@@ -1,43 +1,26 @@
// Central contract re-export point: every legacy API contract import inside
// the Connection package goes through this browser-safe file.
// Types and runtime protocol helpers/bounds come from the apiproxy api/ layer
// (zero Node deps, browser-safe); AbstractApiClient is the client boundary.
// NEVER import the package root: it drags bootHost/cordis into the browser bundle.
// The ./api and ./client subpath exports are the browser-safe channels.
/** Browser-safe Connection protocol and shared application value exports. */
export type {
ApiProxy, HostApi,
ResponseValue,
ModelCatalog, ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
ModelReasoningEffort, ModelSelection,
} from '@deepseek-ai/dsh-host-apiproxy/api'
export type {
RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
ClientRequest, ServerResponse, RpcMessage,
} from '@deepseek-ai/dsh-host-apiproxy/api'
// transportError lives in the apiproxy api layer (beside RpcResult, its
// subject); re-exported here so connection consumers keep one contract
// entry point.
export {
RpcId,
transportError,
} from '@deepseek-ai/dsh-host-apiproxy/api'
export { AbstractApiClient } from '@deepseek-ai/dsh-host-apiproxy/client'
export type { IApiClient } from '@deepseek-ai/dsh-host-apiproxy/client'
ClientRequest,
RpcError,
RpcErrorCode,
RpcMessage,
RpcRequest,
RpcResponse,
RpcResult,
ServerResponse,
} from '../rpc.ts'
export { RpcId, transportError } from '../rpc.ts'
export type { SessionId, SessionEvent } from '@deepseek-ai/dsh-session/types'
export type { MessageId } from '@deepseek-ai/dsh-llm/brand'
export type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm/types'
/** Successful value returned by the connection-generation host handshake. */
export type HostDescription = import('@deepseek-ai/dsh-host-apiproxy/api').ResponseValue<'host.describe'>
import type { RpcResponse, RpcResult } from '@deepseek-ai/dsh-host-apiproxy/api'
import type { RpcResponse, RpcResult } from '../rpc.ts'
/**
* Unwrap a unary response: RpcResponse<T> -> RpcResult<T> (business code only
* cares about the result slot).
* @param response - the unary response.
* @returns its result slot.
* Return the business result carried by a narrow fixture response.
* @param response - fixture response to unwrap.
* @returns the response's business result.
*/
export function resultOf<T>(response: RpcResponse<T>): RpcResult<T> {
return response.result
packages/client/connection/src/client/connection.ts
+5 -19
View File
@@ -1,5 +1,3 @@
import type { HostDescription, IApiClient } from './api.ts'
/** Stable Host facts delivered by one established Remote event generation. */
export interface ConnectionHostInfo {
/** Host account home used only to abbreviate displayed filesystem paths. */
@@ -52,8 +50,8 @@ export type ConnectionState = 'connected' | 'reconnecting'
/** Connection-generation callbacks owned by API Gateway. */
export interface ConnectionSinks {
/** After the generation source is ready and host.describe succeeds, first connect included. */
onConnected?: (description: HostDescription, host: ConnectionHostInfo) => void
/** After the generation source reports ready, first connect included. */
onConnected?: (host: ConnectionHostInfo) => void
/** Coarse state transitions (deduplicated: fires only on change). The initial pre-connect
* span reports nothing — the UI treats "no state yet" as connecting, not as an outage. */
onStateChange?: (state: ConnectionState) => void
@@ -86,7 +84,6 @@ export class ConnectionController {
private readonly config: Required<ConnectionConfig>
constructor(
private readonly api: IApiClient,
private readonly source: ConnectionGenerationSource,
private readonly sinks: ConnectionSinks = {},
config: ConnectionConfig = {},
@@ -173,27 +170,16 @@ export class ConnectionController {
})
try {
// The source reports ready only after its incremental listeners exist;
// describe may complete in parallel, but consumers see neither result
// until both sides of the baseline-plus-increment handshake are ready.
const [description, host] = await Promise.race([
Promise.all([
this.api.host.describe({}, ac.signal),
const host = await Promise.race([
waitForReady(ready, this.config.generationReadyTimeoutMs, ac.signal),
]),
sourceLost,
])
const descriptionResult = description.result
if (!descriptionResult.ok) {
throw new Error(`host.describe failed: ${descriptionResult.error.code}: ${descriptionResult.error.message}`)
}
if (ac.signal.aborted) throw new Error('generation aborted during readiness handshake')
this.attempt = 0
this.emitState('connected')
// A state sink may synchronously stop this controller. Do not publish
// a description for a generation that no longer exists afterward.
// A state sink may synchronously stop this controller.
if (this.isGenerationActive(ac)) {
this.callSink(() => { this.sinks.onConnected?.(descriptionResult.value, host) })
this.callSink(() => { this.sinks.onConnected?.(host) })
}
} catch {
// Transport failure: treat as generation failure, fall through to the shared backoff.
packages/client/connection/src/client/fixture.ts
+32 -95
View File
@@ -33,12 +33,7 @@ import type { CredentialInfo } from '@deepseek-ai/dsh-credentials/types'
import type { DirectoryListing as FixtureDirectoryListing } from '@deepseek-ai/dsh-host-directory-picker/types'
import type { SettingsDescribeValue, SettingsNamespaceView } from '@deepseek-ai/dsh-settings/types'
import { deriveEventMessage, foldSurface } from '@deepseek-ai/dsh-session/surface'
import type {
ApiProxy, ClientRequest,
ModelProviderGroup, ModelSelection, RpcRequest, RpcResponse, RpcResult, ServerResponse,
} from './api.ts'
import type { RequestPayload, ResponseValue, RpcMethodMap } from '@deepseek-ai/dsh-host-apiproxy/api'
import { AbstractApiClient, RpcId } from './api.ts'
import type { RpcResult } from './api.ts'
import { randomUuid } from './random-uuid.ts'
import type {
ClientConnectionRpc, ConnectionRpcFailure, ConnectionRpcResult,
@@ -46,6 +41,26 @@ import type {
const FIXTURE_SESSION_SEARCH_RESULT_LIMIT = 20
interface ModelSelection {
readonly provider: string
readonly model: string
readonly reasoningEffort?: string
}
interface ModelProviderGroup {
readonly id: string
readonly name: string
readonly models: readonly {
readonly id: string
readonly name: string
readonly description?: string
readonly reasoning?: {
readonly efforts: readonly { readonly id: string; readonly name: string; readonly description?: string }[]
readonly defaultEffort?: string
}
}[]
}
/* jscpd:ignore-start -- The standalone fixture mirrors host timing without importing a target implementation. */
function isFixtureTokenDelta(chunk: StreamChunk): boolean {
switch (chunk.type) {
@@ -324,11 +339,6 @@ interface FixtureWorkspace {
updatedAt: string
}
/** The fake carrier mints like a real one (business code never mints). */
function rpcRequest<P>(payload: P): RpcRequest<P> {
return { rpcId: RpcId(randomUuid()), payload }
}
function text(t: string): ContentBlock[] {
return [{ type: 'text', text: t }]
}
@@ -1731,34 +1741,22 @@ class FxInbox<Value> implements StreamConn<Value> {
}
}
/**
* In-memory fake host: fx-alpha carries history and replay scripts; fx-beta is fx-alpha's child session (lineage indent material).
* @param options - fixture branches for empty state and failure timing.
* @returns an ApiProxy backed entirely by in-memory state — no host process, no network.
*/
export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
return createFixtureWorld(options).api
}
/** Both fixture faces over one state graph. */
/** Fixture RPC face over one in-memory state graph. */
export interface FixtureWorld {
/** Legacy unary/stream API the fixture still answers. */
readonly api: ApiProxy
/** Generic Remote caller for the endpoints business services own. */
readonly rpc: ClientConnectionRpc
}
/**
* Build both fixture faces so a caller can drive the Remote endpoints and the
* legacy API against one in-memory state graph.
* Build the fixture RPC face over one in-memory state graph.
* @param options - fixture branches for empty state and failure timing.
* @returns the legacy API face and the Remote RPC face.
* @returns the Remote RPC face.
*/
export function createFixtureFaces(options: FixtureOptions = {}): FixtureWorld {
return createFixtureWorld(options)
}
/** Build the fixture's legacy API and Remote RPC faces over one state graph. */
/** Build the fixture's Remote RPC face over one state graph. */
function createFixtureWorld(options: FixtureOptions): FixtureWorld {
// The resident fixture sessions all carry history, so none of them is blank.
const sessions: FixtureSessionSummary[] = options.empty ? [] : [
@@ -1890,7 +1888,6 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
let fixtureDefaultPreset = 'standard'
const nextTurn = new Map<SessionId, number>([[sid('fx-alpha'), 75]])
let nextSession = 1
let attachedSessions = options.empty ? 0 : 1
// Workspace entities mirroring the host registry: the fixture sessions all
// live under one workspace, whose account carries them in attach order.
const wid = (raw: string): WorkspaceId => raw as WorkspaceId
@@ -2015,10 +2012,6 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
for (const conn of followConns.get(sessionId) ?? []) conn.push(entry)
}
/** OK response echoing the caller's rpcId (contract: responses always backfill, never mint). */
function ok<P, T>(request: RpcRequest<P>, value: T): Promise<RpcResponse<T>> {
return Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value } })
}
function sessionOk<T>(value: T): Promise<ConnectionRpcResult<T>> {
return Promise.resolve({ ok: true, value })
}
@@ -2817,7 +2810,6 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
}
sessions.push(created)
modelSelections.set(created.sessionId, { provider: 'deepseek-official', model: 'deepseek-v4-flash' })
attachedSessions += 1
const emitSession = (): void => {
emitRemote('api-session/added', [created])
}
@@ -3399,20 +3391,6 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
},
}
const api: ApiProxy = {
host: {
describe: request => ok(request, {
version: '0.0.0-fixture', cwd: '/tmp/fixture', attachedSessions, home: FIXTURE_HOME, canOpenPath: true,
}),
},
// Satisfies the ApiProxy contract type only: the browser export button
// hands GET /api/session.export to the native download manager, so this
// stub is never reached through the fixture's dispatch.
downloads: {
sessionLog: () => Promise.resolve(new Response('fixture mode does not serve session export', { status: 404 })),
},
}
const rpc: ClientConnectionRpc = {
call(channel, endpoint, payload, signal) {
if (channel !== '/api') {
@@ -3486,6 +3464,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
case 'credentials/set': return Promise.resolve(credentialRemotes.set(args.ref as string))
case 'credentials/unset': return Promise.resolve(credentialRemotes.unset(args.ref as string))
case 'settings/describe': return Promise.resolve(settingsRemotes.describe())
case 'settings/canOpenAgentPresetDirectory': return Promise.resolve({ ok: true, value: true })
case 'settings/openSettingsDocument': return Promise.resolve(settingsRemotes.openSettingsDocument())
case 'settings/openAgentPresetDirectory': return Promise.resolve(
settingsRemotes.openAgentPresetDirectory(args.agentPreset as string),
@@ -3504,6 +3483,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
case 'session/openWorkspacePath': {
return sessionOk({ opened: true as const })
}
case 'session/canOpenWorkspacePath': return Promise.resolve({ ok: true, value: true })
case 'session/modelCatalog': return Promise.resolve({
ok: true,
value: {
@@ -3610,58 +3590,15 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
}
},
}
return { api, rpc }
return { rpc }
}
/**
* Fixture platform subclass: there is no HTTP at all, so instead of a doFetch transport it
* overrides the legacy protocol-level call virtual to dispatch
* straight into the in-memory ApiProxy while still minting rpcIds, fabricating
* the request/response envelopes, and feeding the same tap as a real carrier. TODO: delete when the fixture
* moves to the isomorphic pipeline (InProcessApiClient over toFetchHandler(fixtureImpl)).
* Build the browser fixture transport from the current page's query switches.
* @returns an in-memory Connection RPC transport.
*/
export class FixtureApiClient extends AbstractApiClient {
private readonly api: ApiProxy
/** Generic Remote caller backed by the same in-memory state as the legacy fixture API. */
readonly rpc: ClientConnectionRpc
constructor() {
super()
const world = createFixtureWorld(fixtureOptionsFromLocation())
this.api = world.api
this.rpc = world.rpc
}
protected doFetch(): Promise<Response> {
throw new Error('FixtureApiClient overrides all protocol paths; doFetch must be unreachable')
}
protected override async callUnary<K extends keyof RpcMethodMap>(
method: K,
payload: RequestPayload<K>,
signal?: AbortSignal,
): Promise<RpcResponse<ResponseValue<K>>> {
void signal
const request = rpcRequest(payload)
const full: ClientRequest = { type: 'client-request', rpcId: request.rpcId, method, payload }
this.onEnvelope(full)
const response = await this.dispatch(
method,
request as RpcRequest<never>,
) as RpcResponse<ResponseValue<K>>
const fullResponse: ServerResponse = { type: 'server-response', rpcId: response.rpcId, result: response.result }
this.onEnvelope(fullResponse)
return response
}
/** Method-key dispatch into the in-memory contract impl (a real carrier routes by URL path instead). */
private dispatch(
_method: keyof RpcMethodMap,
request: RpcRequest<never>,
): Promise<RpcResponse<unknown>> {
return this.api.host.describe(request)
}
export function createFixtureConnectionRpc(): ClientConnectionRpc {
return createFixtureWorld(fixtureOptionsFromLocation()).rpc
}
/** Browser query mapping; direct unit callers pass FixtureOptions explicitly. */
packages/client/connection/src/client/index.ts
+9 -58
View File
@@ -3,7 +3,6 @@
* the shared API client, and lets API Gateway own the connection loop.
*/
import type { Context } from '@deepseek-ai/cordis'
import type { HostDescription, IApiClient } from './api.ts'
import {
ConnectionController,
type ConnectionConfig,
@@ -11,8 +10,7 @@ import {
type ConnectionGenerationSource,
type ConnectionSinks,
} from './connection.ts'
import { FixtureApiClient } from './fixture.ts'
import { WebApiClient } from './web-api-client.ts'
import { createFixtureConnectionRpc } from './fixture.ts'
import { createWebConnectionRpc, type RpcFetch, type RpcStreamOpen } from './rpc.ts'
import { isLoopbackHostname } from '../loopback-hostname.ts'
import type { ClientConnectionRpc } from '../rpc.ts'
@@ -28,18 +26,15 @@ declare module '@deepseek-ai/cordis' {
}
}
// ---- Contract re-exports (browser-safe apiproxy channels + core types) ----
// ---- Browser-safe protocol and shared value re-exports ----
export type {
ApiProxy, HostApi,
ModelCatalog, ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
MessageId, ModelReasoningEffort, ModelSelection,
MessageId,
RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
ClientRequest, ServerResponse, RpcMessage,
HostDescription, IApiClient, SessionId, SessionEvent, ContentBlock, StreamChunk,
SessionId, SessionEvent, ContentBlock, StreamChunk,
} from './api.ts'
export {
RpcId,
AbstractApiClient,
transportError,
} from './api.ts'
@@ -58,14 +53,6 @@ export type {
} from '../rpc.ts'
export type { RpcFetch } from './rpc.ts'
/** Observable Host description published by each completed connection handshake. */
export interface HostDescriptionSource {
/** Latest connected-generation description; absent before connect and while reconnecting. */
getSnapshot(): HostDescription | undefined
/** Subscribe to description replacement and connection loss. */
subscribe(listener: () => void): () => void
}
/** Observable identity and Host facts for the active connection generation. */
export interface ConnectionGenerationState {
/** Active generation, or undefined before readiness and while reconnecting. */
@@ -84,8 +71,6 @@ export const inject: string[] = []
* provides both halves here instead of forking this plugin.
*/
export interface ClientTransportHooks {
/** Build the API carrier: unary calls plus the two downstream event streams. */
createApiClient(): IApiClient
/** Transport for generic unary RPC channels (the Typert gateway). */
fetch: RpcFetch
/** Worker-local Gateway stream carrier; absent when the page uses the Gateway WebSocket. */
@@ -118,16 +103,12 @@ interface ClientTransportGlobal {
* Connection stays independent of downstream domain state.
*/
export interface ConnectionHandle {
/** Shared api client (fixture or real, decided at boot from the page URL). */
readonly api: IApiClient
/**
* Whether the privileged surface is reachable: the page authority is
* loopback, the transport declares the page owns the Host
* ({@link ClientTransportHooks.ownsHost}), or the context is not a browser.
*/
readonly isLoopback: boolean
/** Generation-scoped Host facts, including the account home and native path-open capability. */
readonly hostDescription: HostDescriptionSource
/** Current Remote event generation and the Host facts carried by its opening frame. */
readonly generation: ConnectionGenerationState
/** Generic logical RPC channels over the same Connection transport. */
@@ -162,28 +143,14 @@ interface ConnectionOwner {
export function apply(ctx: Context): void {
const pageLocation = typeof location === 'undefined' ? undefined : location
const fixture = pageLocation !== undefined && new URLSearchParams(pageLocation.search).has('fixture')
const fixtureClient = fixture ? new FixtureApiClient() : undefined
const fixtureRpc = fixture ? createFixtureConnectionRpc() : undefined
const transport = (globalThis as ClientTransportGlobal).__DSH_TRANSPORT__
const api: IApiClient = fixtureClient ?? transport?.createApiClient() ?? new WebApiClient()
const rpc = fixtureClient?.rpc ?? createWebConnectionRpc(transport?.fetch, transport?.openStream)
const rpc = fixtureRpc ?? createWebConnectionRpc(transport?.fetch, transport?.openStream)
let generationSource: ConnectionGenerationSource | undefined
let owner: ConnectionOwner | undefined
let generationId = 0
let generation: ConnectionGeneration | undefined
const generationListeners = new Set<() => void>()
let description: HostDescription | undefined
const descriptionListeners = new Set<() => void>()
const publishDescription = (next: HostDescription | undefined): void => {
if (Object.is(description, next)) return
description = next
for (const listener of [...descriptionListeners]) {
try {
listener()
} catch (error) {
console.error('[connection] host-description listener threw:', error)
}
}
}
const publishGeneration = (next: ConnectionGeneration | undefined): void => {
if (Object.is(generation, next)) return
generation = next
@@ -200,18 +167,9 @@ export function apply(ctx: Context): void {
owner = undefined
current.controller.stop()
publishGeneration(undefined)
publishDescription(undefined)
}
const handle: ConnectionHandle = {
api,
isLoopback: transport?.ownsHost === true || pageLocation === undefined || isLoopbackHostname(pageLocation.hostname),
hostDescription: {
getSnapshot: () => description,
subscribe: (listener) => {
descriptionListeners.add(listener)
return () => { descriptionListeners.delete(listener) }
},
},
generation: {
getSnapshot: () => generation,
subscribe: (listener) => {
@@ -238,24 +196,17 @@ export function apply(ctx: Context): void {
if (source === undefined) throw new Error('connection: no generation source is registered')
const token = {}
const ownsGeneration = (): boolean => owner?.token === token
const controller = new ConnectionController(api, source, {
const controller = new ConnectionController(source, {
...sinks,
onConnected: (next, host) => {
onConnected: (host) => {
const nextGeneration = { id: ++generationId, host }
publishGeneration(nextGeneration)
if (!ownsGeneration() || !Object.is(generation, nextGeneration)) return
publishDescription(next)
// A description subscriber may synchronously stop the loop. In that
// case publishDescription(undefined) has already retracted this
// generation, so do not leak its stale connected notification to
// the consumer sink afterward.
if (!ownsGeneration() || !Object.is(description, next)) return
sinks.onConnected?.(next, host)
sinks.onConnected?.(host)
},
onStateChange: (state) => {
if (state === 'reconnecting') {
publishGeneration(undefined)
publishDescription(undefined)
}
if (!ownsGeneration()) return
sinks.onStateChange?.(state)
packages/client/connection/src/client/rpc.ts
+1 -1
View File
@@ -4,7 +4,7 @@ import {
RpcId,
type ClientRequest,
type RpcId as RpcIdType,
} from '@deepseek-ai/dsh-host-apiproxy/api'
} from '../rpc.ts'
import type { ClientConnectionRpc, ConnectionRpcResult } from '../rpc.ts'
import { randomUuid } from './random-uuid.ts'
packages/client/connection/src/client/web-api-client.ts
-10
View File
@@ -1,10 +0,0 @@
/** Browser API carrier for unary HTTP calls. */
import { AbstractApiClient } from './api.ts'
/** Browser platform subclass supplying fetch for unary calls. */
export class WebApiClient extends AbstractApiClient {
protected doFetch(input: URL, init?: RequestInit): Promise<Response> {
return globalThis.fetch(input, init)
}
}
packages/client/connection/src/index.ts
+19 -11
View File
@@ -5,7 +5,6 @@ import type {} from '@deepseek-ai/dsh-attachment'
import type {} from '@deepseek-ai/dsh-credentials'
// Activates the webServer Context merge used below.
import type { WebRoute } from '@deepseek-ai/dsh-host-webserver'
import { toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
import { API_PATH } from './api-path.ts'
import { bridge, DEFAULT_MAX_REQUEST_BODY_BYTES } from './http-bridge.ts'
import { assertTrustedAuthority } from './api-request-trust.ts'
@@ -14,6 +13,7 @@ import { HostConnectionService } from './rpc-host.ts'
export type {
ConnectionFetchMethod,
ConnectionFetchHandler,
ConnectionFetchRoute,
ConnectionIndexRequest,
ConnectionIndexResponse,
@@ -23,10 +23,22 @@ export type {
ConnectionRequestRejection,
ConnectionRpcResult,
ConnectionTrustRequest,
ClientRequest,
HostConnectionHandle,
HostConnectionFetch,
HostConnectionRpc,
RpcMessage,
ServerResponse,
} from './rpc.ts'
export { RpcId, transportError } from './rpc.ts'
export {
clientRequestSchema,
rpcErrorSchema,
rpcIdSchema,
rpcMessageSchema,
rpcResultSchema,
serverResponseSchema,
} from './rpc-schema.ts'
export { HostConnectionService } from './rpc-host.ts'
export { API_PATH } from './api-path.ts'
@@ -51,7 +63,7 @@ function assertImageBodyCapacity(ctx: Context, maxRequestBodyBytes: number): voi
}
}
/** Services required before providing Connection; API Proxy is an optional `/api` fallback. */
/** Services required before providing Connection. */
export const inject = ['webServer', 'credentials']
/** Plugin config: the deployment's non-loopback serving authorities. */
@@ -92,19 +104,13 @@ export async function apply(ctx: Context, config?: ConnectionConfig): Promise<vo
// Config boundary: a malformed entry fails the load loudly here rather than
// silently authorizing its hostname prefix at request time.
for (const entry of trustedHosts) assertTrustedAuthority(entry)
if (ctx.get('apiProxy') !== undefined) assertImageBodyCapacity(ctx, maxRequestBodyBytes)
assertImageBodyCapacity(ctx, maxRequestBodyBytes)
const connection = new HostConnectionService(
ctx,
trustedHosts,
await BrowserAuth.create(ctx.root, ctx.credentials, cookieMaxAgeDays),
)
const fetchHandler = connection.createSharedFetchHandler(API_PATH, {
async fetch(request) {
const apiProxy = ctx.get('apiProxy')
if (apiProxy === undefined) return new Response('not found', { status: 404 })
return await toFetchHandler(apiProxy).fetch(request)
},
})
const fetchHandler = connection.createSharedFetchHandler(API_PATH)
const route: WebRoute = {
kind: 'prefix',
path: API_PATH,
@@ -119,5 +125,7 @@ export async function apply(ctx: Context, config?: ConnectionConfig): Promise<vo
},
}
ctx.effect(() => ctx.webServer.register(route), 'client-connection: /api route')
ctx.inject(['apiProxy'], (apiCtx) => { assertImageBodyCapacity(apiCtx, maxRequestBodyBytes) })
ctx.inject(['attachments'], (attachmentCtx) => {
assertImageBodyCapacity(attachmentCtx, maxRequestBodyBytes)
})
}
packages/client/connection/src/invariant.ts
+2 -2
View File
@@ -18,8 +18,8 @@ export const inject = ['invariants']
* No runtime invariant: browser-session verification reads the credential
* record asynchronously at the request that authorizes work, while the
* credentials companion owns record commit-event lifetime. Stream/reconnect
* sequencing is exercised directly by behavior specs, rpcId round-trip
* discipline belongs to apiproxy, and route register/dispose symmetry is
* sequencing and rpcId round-trip discipline are exercised directly by
* behavior specs, and route register/dispose symmetry is
* audited by the webserver companion.
*/
const install: InvariantInstaller = () => {}
packages/client/connection/src/rpc-host.ts
+10 -17
View File
@@ -3,13 +3,11 @@
import { Context, Service } from '@deepseek-ai/cordis'
import type { WebRoute } from '@deepseek-ai/dsh-host-webserver'
import {
clientRequestSchema,
RpcId,
type ClientRequest,
type RpcError,
type RpcErrorDetailsMap,
type RpcId as RpcIdType,
} from '@deepseek-ai/dsh-host-apiproxy/api'
} from './rpc.ts'
import { clientRequestSchema } from './rpc-schema.ts'
import { bridge, type FetchHandler } from './http-bridge.ts'
import { isTrustedApiRequest } from './api-request-trust.ts'
import { API_PATH } from './api-path.ts'
@@ -18,8 +16,10 @@ import type {
ConnectionIndexRequest,
ConnectionIndexResponse,
ConnectionFetchRoute,
ConnectionFetchHandler,
HostConnectionFetch,
ConnectionRpcEndpointMatcher,
ConnectionRpcFailure,
ConnectionRpcHandler,
ConnectionRpcResult,
ConnectionRequestRejection,
@@ -109,15 +109,13 @@ export class HostConnectionService extends Service implements HostConnectionHand
}
/**
* Compose one shared-channel Fetch handler from its interceptor and fallback.
* Compose one shared-channel Fetch handler from exact routes and its interceptor.
* @param channel - shared channel mounted by Connection.
* @param fallback - handler for endpoints not claimed by the interceptor.
* @returns Fetch handler that selects exactly one target for each request.
* @returns Fetch handler that selects one owner or returns 404.
*/
createSharedFetchHandler(
channel: '/api',
fallback: FetchHandler,
): FetchHandler {
): ConnectionFetchHandler {
return {
fetch: (request) => {
const pathname = new URL(request.url).pathname
@@ -126,7 +124,7 @@ export class HostConnectionService extends Service implements HostConnectionHand
const endpoint = endpointFromPath(channel, pathname)
const interceptor = this.interceptors.get(channel)
if (endpoint === undefined || interceptor === undefined || !interceptor.matches(endpoint)) {
return fallback.fetch(request)
return Promise.resolve(new Response('not found', { status: 404 }))
}
return interceptor.fetchHandler.fetch(request)
},
@@ -248,7 +246,7 @@ function rpcFetchHandler(
}
}
function invalidEnvelopeResponse(body: unknown, issues: RpcErrorDetailsMap['bad-request']['issues']): Response {
function invalidEnvelopeResponse(body: unknown, issues: readonly object[]): Response {
const rawId = (body as { rpcId?: unknown } | null)?.rpcId
const rpcId = typeof rawId === 'string' ? RpcId(rawId) : INVALID_REQUEST_RPC_ID
return errorResponse(rpcId, {
@@ -269,7 +267,7 @@ function endpointFromPath(channel: string, pathname: string): string | undefined
return endpoint
}
function errorResponse(rpcId: RpcIdType, error: RpcError): Response {
function errorResponse(rpcId: RpcIdType, error: ConnectionRpcFailure): Response {
return fullResponse(rpcId, { ok: false, error })
}
@@ -295,9 +293,4 @@ function assertFetchRoute(route: ConnectionFetchRoute): void {
if (methods.size !== route.methods.length) {
throw new Error(`connection: exact Fetch route ${JSON.stringify(route.path)} repeats a method`)
}
for (const method of methods) {
if (method !== 'GET' && method !== 'HEAD') {
throw new Error(`connection: exact Fetch route ${JSON.stringify(route.path)} has unsupported method ${JSON.stringify(method)}`)
}
}
}
packages/client/connection/src/rpc-schema.ts
+53
View File
@@ -0,0 +1,53 @@
/** Runtime validation for Connection RPC envelopes. */
import { z } from 'zod'
import type { ClientRequest, RpcId, RpcMessage, ServerResponse } from './rpc.ts'
/** Correlation id after wire validation. */
export const rpcIdSchema = z.string() as unknown as z.ZodType<RpcId>
/** Generic endpoint failure carried in a response envelope. */
export const rpcErrorSchema = z.object({
code: z.string(),
message: z.string(),
details: z.record(z.string(), z.unknown()),
})
/**
* Build the result parser for one endpoint value parser.
* @param value - endpoint-owned success-value parser.
* @returns parser for either a success value or generic failure.
*/
export function rpcResultSchema<T>(value: z.ZodType<T>): z.ZodType<{
readonly ok: true
readonly value: T
} | {
readonly ok: false
readonly error: z.infer<typeof rpcErrorSchema>
}> {
return z.union([
z.object({ ok: z.literal(true), value }),
z.object({ ok: z.literal(false), error: rpcErrorSchema }),
])
}
/** Client request envelope; endpoint payload validation belongs to its owner. */
export const clientRequestSchema = z.object({
type: z.literal('client-request'),
rpcId: rpcIdSchema,
method: z.string(),
payload: z.unknown(),
}) as z.ZodType<ClientRequest>
/** Server response envelope; endpoint value validation belongs to its caller. */
export const serverResponseSchema = z.object({
type: z.literal('server-response'),
rpcId: rpcIdSchema,
result: rpcResultSchema(z.unknown().optional()),
}) as z.ZodType<ServerResponse>
/** Either Connection RPC envelope direction. */
export const rpcMessageSchema = z.discriminatedUnion('type', [
clientRequestSchema as unknown as z.ZodObject<z.ZodRawShape>,
serverResponseSchema as unknown as z.ZodObject<z.ZodRawShape>,
]) as unknown as z.ZodType<RpcMessage>
packages/client/connection/src/rpc.ts
+107
View File
@@ -1,5 +1,20 @@
/** Generic unary RPC contracts shared by the Host and Client Connection halves. */
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
/** Correlation id minted by a caller and echoed by the Connection response. */
export type RpcId = Branded<'rpc-id'>
/**
* Brand one validated string as a Connection correlation id.
* @param id - validated wire identity.
* @returns the same string with the correlation-id brand.
*/
export function RpcId(id: string): RpcId {
return id as RpcId
}
/** Carrier-neutral failure returned by one logical RPC endpoint. */
export interface ConnectionRpcFailure {
readonly code: string
@@ -12,6 +27,81 @@ export type ConnectionRpcResult<T> =
| { readonly ok: true; readonly value: T }
| { readonly ok: false; readonly error: ConnectionRpcFailure }
/** Typed failure details used by Client Session adapters. */
export interface RpcErrorDetailsMap {
'bad-request': { issues: object[] }
'cancelled': {}
'session-not-found': { sessionId: SessionId }
'invalid-time-zone': { value: string }
'agent-preset-read-only': { agentPreset: string; reason: string }
'agent-preset-locked': { sessionId: SessionId; agentPreset: string }
'agent-preset-not-found': { agentPreset: string; available: readonly string[] }
'agent-preset-invalid': { agentPreset: string; reason: string }
'agent-busy': { reason: string }
'internal': {}
}
/** Error codes used by Client Session adapters. */
export type RpcErrorCode = keyof RpcErrorDetailsMap
/** Typed failure used by Client Session adapters. */
export type RpcError = {
[Code in RpcErrorCode]: {
readonly code: Code
readonly message: string
readonly details: RpcErrorDetailsMap[Code]
}
}[RpcErrorCode]
/** Historical short name for a generic Connection result. */
export type RpcResult<T> = ConnectionRpcResult<T>
/**
* Convert a rejected transport operation into a generic failure result.
* @param error - rejected transport value.
* @returns an `internal` failure preserving the available message.
*/
export function transportError<T>(error: unknown): RpcResult<T> {
return {
ok: false,
error: {
code: 'internal',
message: error instanceof Error ? error.message : String(error),
details: {},
},
}
}
/** Narrow request form used by direct fixture adapters. */
export interface RpcRequest<P> {
readonly rpcId: RpcId
readonly payload: P
}
/** Narrow response form used by direct fixture adapters. */
export interface RpcResponse<T> {
readonly rpcId: RpcId
readonly result: RpcResult<T>
}
/** Full request envelope carried by Connection RPC transports. */
export interface ClientRequest {
readonly type: 'client-request'
readonly rpcId: RpcId
readonly method: string
readonly payload: unknown
}
/** Full response envelope carried by Connection RPC transports. */
export interface ServerResponse {
readonly type: 'server-response'
readonly rpcId: RpcId
readonly result: ConnectionRpcResult<unknown>
}
/** Complete Connection RPC envelope union. */
export type RpcMessage = ClientRequest | ServerResponse
/** HTTP request facts consumed by browser trust and authentication. */
export interface ConnectionTrustRequest {
/** Request headers supplied by either the Fetch or node:http representation. */
@@ -100,6 +190,13 @@ export interface HostConnectionHandle {
/** Exact Fetch routes for streaming or browser-native responses. */
readonly fetch: HostConnectionFetch
/**
* Compose exact Fetch routes and the shared-channel RPC interceptor.
* @param channel - shared channel mounted by Connection.
* @returns Fetch handler for trusted, authenticated requests.
*/
createSharedFetchHandler(channel: '/api'): ConnectionFetchHandler
/**
* Apply Connection's Host/Origin checks and browser authentication to
* another Web route.
@@ -124,6 +221,16 @@ export interface HostConnectionHandle {
authenticatedUrl(baseUrl: string): string
}
/** Transport-independent Fetch handler used by HTTP and worker carriers. */
export interface ConnectionFetchHandler {
/**
* Dispatch one already-authenticated request.
* @param request - Fetch request below the shared channel.
* @returns the registered response or a 404 response.
*/
fetch(request: Request): Promise<Response>
}
/** Client caller for logical RPC channels carried by the current transport. */
export interface ClientConnectionRpc {
/**
packages/client/connection/tests/client-apply.client.spec.ts
+49 -68
View File
@@ -10,8 +10,6 @@ import {
type ConnectionGenerationSource,
type ConnectionHandle,
} from '../src/client/index.ts'
import { FixtureApiClient } from '../src/client/fixture.ts'
import { WebApiClient } from '../src/client/web-api-client.ts'
type Win = {
location?: { hostname: string; search: string; origin?: string }
@@ -61,20 +59,22 @@ async function mount(): Promise<ConnectionHandle> {
}
describe('connection client apply', () => {
it('mounts ctx.connection with the real client when no ?fixture switch is present', async () => {
it('treats a runtime without browser location as local', async () => {
delete (globalThis as Win).location
expect((await mount()).isLoopback).toBe(true)
})
it('mounts ctx.connection and identifies a loopback page', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '' }
const handle = await mount()
expect(handle.api).toBeInstanceOf(WebApiClient)
expect(handle.isLoopback).toBe(true)
})
it('selects the fixture client under ?fixture (and with no location at all stays real)', async () => {
it('selects the fixture RPC transport under ?fixture', async () => {
;(globalThis as Win).location = { hostname: '127.0.0.1', search: '?fixture' }
expect((await mount()).api).toBeInstanceOf(FixtureApiClient)
delete (globalThis as Win).location
const handle = await mount()
expect(handle.api).toBeInstanceOf(WebApiClient)
expect(handle.isLoopback).toBe(true)
await expect(handle.rpc.call('/api', 'settings/describe', { args: {} }))
.resolves.toMatchObject({ ok: true })
})
it('reports non-loopback page authority through the connection handle', async () => {
@@ -98,10 +98,10 @@ describe('connection client apply', () => {
const loop = handle.start({})
await vi.waitFor(() => {
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
})
unregisterSecond()
expect(handle.hostDescription.getSnapshot()).toBeUndefined()
expect(handle.generation.getSnapshot()).toBeUndefined()
loop.stop()
})
@@ -110,26 +110,26 @@ describe('connection client apply', () => {
const handle = await mount()
installGeneration(handle)
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
const descriptions: Array<boolean | undefined> = []
const stopThrowing = handle.hostDescription.subscribe(() => { throw new Error('subscriber bug') })
const stopDescription = handle.hostDescription.subscribe(() => {
descriptions.push(handle.hostDescription.getSnapshot()?.canOpenPath)
const generations: Array<string | undefined> = []
const stopThrowing = handle.generation.subscribe(() => { throw new Error('subscriber bug') })
const stopGeneration = handle.generation.subscribe(() => {
generations.push(handle.generation.getSnapshot()?.host.home)
})
expect(handle.hostDescription.getSnapshot()).toBeUndefined()
expect(handle.generation.getSnapshot()).toBeUndefined()
// config omitted: the `config ?? {}` default arm is part of the surface.
let connected = 0
const loop = handle.start({ onConnected: () => { connected++ } })
expect(() => handle.start({})).toThrow(/already owned by another consumer/)
await vi.waitFor(() => {
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
})
loop.stop() // teardown must not throw; the fixture streams abort quietly
expect(handle.hostDescription.getSnapshot()).toBeUndefined()
expect(descriptions).toEqual([true, undefined])
expect(handle.generation.getSnapshot()).toBeUndefined()
expect(generations).toEqual(['/h', undefined])
expect(connected).toBe(1)
expect(errorSpy).toHaveBeenCalledTimes(2)
stopThrowing()
stopDescription()
stopGeneration()
errorSpy.mockRestore()
})
@@ -140,87 +140,87 @@ describe('connection client apply', () => {
const first = handle.start({})
await vi.waitFor(() => {
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
})
first.stop()
expect(handle.hostDescription.getSnapshot()).toBeUndefined()
expect(handle.generation.getSnapshot()).toBeUndefined()
const second = handle.start({})
await vi.waitFor(() => {
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
})
first.stop()
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
second.stop()
generation.end()
})
it('does not announce a generation synchronously stopped by a description subscriber', async () => {
it('does not announce a generation synchronously stopped by a generation subscriber', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
const handle = await mount()
installGeneration(handle)
const owner: { loop?: ReturnType<ConnectionHandle['start']> } = {}
let sawDescription = false
const stopDescription = handle.hostDescription.subscribe(() => {
if (handle.hostDescription.getSnapshot() === undefined) return
sawDescription = true
let sawGeneration = false
const stopGeneration = handle.generation.subscribe(() => {
if (handle.generation.getSnapshot() === undefined) return
sawGeneration = true
owner.loop?.stop()
})
const connected = vi.fn()
const loop = handle.start({ onConnected: connected })
owner.loop = loop
try {
await vi.waitFor(() => { expect(sawDescription).toBe(true) })
expect(handle.hostDescription.getSnapshot()).toBeUndefined()
await vi.waitFor(() => { expect(sawGeneration).toBe(true) })
expect(handle.generation.getSnapshot()).toBeUndefined()
expect(connected).not.toHaveBeenCalled()
} finally {
stopDescription()
stopGeneration()
loop.stop()
}
})
it('retracts the host description while reconnecting and republishes the next generation', async () => {
it('retracts the generation while reconnecting and publishes the next generation', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
const handle = await mount()
const generation = installGeneration(handle)
const descriptions: Array<boolean | undefined> = []
const reconnectSnapshots: Array<boolean | undefined> = []
const stopDescription = handle.hostDescription.subscribe(() => {
descriptions.push(handle.hostDescription.getSnapshot()?.canOpenPath)
const generations: Array<string | undefined> = []
const reconnectSnapshots: Array<string | undefined> = []
const stopGeneration = handle.generation.subscribe(() => {
generations.push(handle.generation.getSnapshot()?.host.home)
})
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const loop = handle.start({
onStateChange: (state) => {
if (state === 'reconnecting') {
reconnectSnapshots.push(handle.hostDescription.getSnapshot()?.canOpenPath)
reconnectSnapshots.push(handle.generation.getSnapshot()?.host.home)
}
},
}, { backoffBaseMs: 10, backoffFactor: 1, backoffMaxMs: 10, generationReadyTimeoutMs: 500 })
try {
await vi.waitFor(() => {
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
})
generation.end()
await vi.waitFor(() => { expect(reconnectSnapshots).toEqual([undefined]) })
await vi.waitFor(() => { expect(descriptions).toEqual([true, undefined, true]) })
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
await vi.waitFor(() => { expect(generations).toEqual(['/h', undefined, '/h']) })
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
} finally {
stopDescription()
stopGeneration()
loop.stop()
warnSpy.mockRestore()
}
})
it('does not announce reconnecting after a description subscriber stops the loop', async () => {
it('does not announce reconnecting after a generation subscriber stops the loop', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
const handle = await mount()
const generation = installGeneration(handle)
const owner: { loop?: ReturnType<ConnectionHandle['start']> } = {}
let stoppedOnRetraction = false
const stopDescription = handle.hostDescription.subscribe(() => {
if (handle.hostDescription.getSnapshot() !== undefined || owner.loop === undefined) return
const stopGeneration = handle.generation.subscribe(() => {
if (handle.generation.getSnapshot() !== undefined || owner.loop === undefined) return
stoppedOnRetraction = true
owner.loop.stop()
})
@@ -232,38 +232,20 @@ describe('connection client apply', () => {
owner.loop = loop
try {
await vi.waitFor(() => {
expect(handle.hostDescription.getSnapshot()?.canOpenPath).toBe(true)
expect(handle.generation.getSnapshot()?.host.home).toBe('/h')
})
generation.end()
await vi.waitFor(() => { expect(stoppedOnRetraction).toBe(true) })
expect(handle.hostDescription.getSnapshot()).toBeUndefined()
expect(handle.generation.getSnapshot()).toBeUndefined()
expect(states).toEqual(['connected'])
} finally {
stopDescription()
stopGeneration()
loop.stop()
warnSpy.mockRestore()
}
})
it('WebApiClient keeps unary calls on globalThis.fetch', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '' }
const handle = await mount()
const original = globalThis.fetch
const seen: string[] = []
globalThis.fetch = (input: URL | RequestInfo) => {
seen.push(typeof input === 'string' ? input : input instanceof URL ? input.href : input.url)
return Promise.resolve(new Response('{}', { status: 200 }))
}
try {
// Schema rejection is fine — the transport hop is the assertion.
await (handle.api as WebApiClient).host.describe({}).catch(() => undefined)
} finally {
globalThis.fetch = original
}
expect(seen.some(u => u.includes('/api/host.describe'))).toBe(true)
})
it('carries RPC calls without requiring secure-context randomUUID', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '' }
vi.stubGlobal('crypto', {
@@ -311,7 +293,6 @@ describe('connection client apply', () => {
})(),
)
;(globalThis as Win).__DSH_TRANSPORT__ = {
createApiClient: () => new FixtureApiClient(),
fetch: vi.fn<ClientTransportHooks['fetch']>(),
openStream,
ownsHost: true,
@@ -428,7 +409,7 @@ describe('connection client apply', () => {
}
})
it('carries Goal Remotes over the same state as the client-only fixture API', async () => {
it('carries Goal Remotes over the client-only fixture state', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
const handle = await mount()
const created = await handle.rpc.call('/api', 'goals/create', {
packages/client/connection/tests/connection.client.spec.ts
+103 -142
View File
@@ -1,117 +1,52 @@
/**
* ConnectionController: strict readiness handshake (describe + incremental
* source ready), generation
* abort on loss, backoff reconnection, state transitions, and sink-exception
* isolation. Real (short) timers — the timeout and backoff are configurable,
* so tests run them at millisecond scale.
*/
/** Connection generation readiness, loss, retry, and sink isolation. */
import { describe, expect, it, vi } from 'vitest'
import type { ConnectionState } from '../src/client/connection.ts'
import type { ConnectionGenerationSource, ConnectionState } from '../src/client/connection.ts'
import { ConnectionController } from '../src/client/connection.ts'
import { FakeApiClient, deferred, ok } from './fake-api.client.ts'
import { FakeGenerationSource } from './fake-generation.client.ts'
const FAST = { backoffBaseMs: 10, backoffFactor: 1, backoffMaxMs: 10, generationReadyTimeoutMs: 500 }
describe('connection lifecycle', () => {
it('announces connected after describe plus generation readiness', async () => {
const api = new FakeApiClient()
const descriptions: boolean[] = []
let connected = 0
const controller = new ConnectionController(api, api.generation, {
onConnected: (description) => {
connected++
descriptions.push(description.canOpenPath)
},
it('announces connected with the Host facts from generation readiness', async () => {
const source = new FakeGenerationSource()
const homes: string[] = []
const controller = new ConnectionController(source.source, {
onConnected: (host) => { homes.push(host.home) },
}, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(connected).toBe(1) })
expect(api.callsOf('host.describe')).toHaveLength(1)
expect(descriptions).toEqual([true])
await vi.waitFor(() => { expect(homes).toEqual(['/h']) })
} finally {
controller.stop()
}
})
it('reconnects with a fresh generation when its source fails, and stop() ends the loop', async () => {
const api = new FakeApiClient()
const source = new FakeGenerationSource()
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(api, api.generation, { onConnected: () => { connected++ } }, FAST)
const controller = new ConnectionController(source.source, { onConnected: () => { connected++ } }, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(connected).toBe(1) })
api.failStreams(new Error('stream torn'))
await vi.waitFor(() => { expect(connected).toBe(2) }) // new generation after backoff
expect(api.openGenerationCount).toBe(1)
source.fail(new Error('stream torn'))
await vi.waitFor(() => { expect(connected).toBe(2) })
expect(source.activeCount).toBe(1)
} finally {
controller.stop()
warnSpy.mockRestore()
}
// stop() aborts the live generation and no reconnect follows.
await vi.waitFor(() => { expect(api.openGenerationCount).toBe(0) })
await vi.waitFor(() => { expect(source.activeCount).toBe(0) })
await new Promise(resolve => setTimeout(resolve, 40))
expect(api.openGenerationCount).toBe(0)
})
it('treats describe failure as generation failure and retries', async () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
let describeCalls = 0
api.onDescribe = () => {
describeCalls++
return describeCalls === 1 ? Promise.reject(new Error('host down')) : gate.promise
}
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(api, api.generation, { onConnected: () => { connected++ } }, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(describeCalls).toBe(2) }) // retried after backoff
expect(connected).toBe(0) // never announced during the failed generation
gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, home: '/h', canOpenPath: true }))
await vi.waitFor(() => { expect(connected).toBe(1) })
} finally {
controller.stop()
warnSpy.mockRestore()
}
})
it('treats a host.describe business error as generation failure', async () => {
const api = new FakeApiClient()
let describeCalls = 0
api.onDescribe = () => {
describeCalls += 1
if (describeCalls === 1) {
return Promise.resolve({
rpcId: 'bad-describe' as never,
result: {
ok: false as const,
error: { code: 'internal' as const, message: 'not ready', details: {} },
},
})
}
return Promise.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, home: '/h', canOpenPath: true }))
}
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(api, api.generation, { onConnected: () => { connected++ } }, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(describeCalls).toBe(2) })
await vi.waitFor(() => { expect(connected).toBe(1) })
} finally {
controller.stop()
warnSpy.mockRestore()
}
expect(source.activeCount).toBe(0)
})
it('isolates a connected sink exception from the generation', async () => {
const api = new FakeApiClient()
const source = new FakeGenerationSource()
let connected = 0
const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
const controller = new ConnectionController(api, api.generation, {
const controller = new ConnectionController(source.source, {
onConnected: () => {
connected++
throw new Error('business layer bug')
@@ -120,7 +55,7 @@ describe('connection lifecycle', () => {
controller.start()
try {
await vi.waitFor(() => { expect(connected).toBe(1) })
expect(api.openGenerationCount).toBe(1)
expect(source.activeCount).toBe(1)
expect(errorSpy).toHaveBeenCalledWith('[connection] connection sink threw:', expect.any(Error))
} finally {
controller.stop()
@@ -128,47 +63,75 @@ describe('connection lifecycle', () => {
}
})
it('holds onConnected until the incremental source is ready after describe succeeds', async () => {
const api = new FakeApiClient()
api.holdGenerationReady = true
it('holds onConnected until the incremental source reports ready', async () => {
const source = new FakeGenerationSource()
source.holdReady = true
let connected = 0
const controller = new ConnectionController(api, api.generation, { onConnected: () => { connected++ } }, FAST)
const controller = new ConnectionController(source.source, { onConnected: () => { connected++ } }, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(api.callsOf('host.describe')).toHaveLength(1) })
await vi.waitFor(() => { expect(source.activeCount).toBe(1) })
await new Promise(resolve => setTimeout(resolve, 30))
expect(connected).toBe(0) // describe alone must not announce
api.releaseGenerationReady()
expect(connected).toBe(0)
source.releaseReady()
await vi.waitFor(() => { expect(connected).toBe(1) })
} finally {
controller.stop()
}
})
it('rejects a generation whose source ends during readiness and retries', async () => {
const api = new FakeApiClient()
const firstDescribe = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
let describeCalls = 0
api.onDescribe = () => {
describeCalls++
return describeCalls === 1
? firstDescribe.promise
: Promise.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, home: '/h', canOpenPath: true }))
it('accepts only the first readiness report from one generation', async () => {
const homes: string[] = []
const source: ConnectionGenerationSource = (signal, ready) => {
ready({ home: '/first' })
ready({ home: '/duplicate' })
return new Promise<void>((resolve) => {
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
}
const controller = new ConnectionController(source, {
onConnected: (host) => { homes.push(host.home) },
}, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(homes).toEqual(['/first']) })
} finally {
controller.stop()
}
})
it('does not announce readiness after a stop queued from the ready callback', async () => {
const owner: { controller?: ConnectionController } = {}
let sourceCalls = 0
const connected = vi.fn()
const source: ConnectionGenerationSource = (signal, ready) => new Promise<void>((resolve) => {
sourceCalls++
ready({ home: '/h' })
queueMicrotask(() => { owner.controller?.stop() })
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
const controller = new ConnectionController(source, { onConnected: connected }, FAST)
owner.controller = controller
controller.start()
await vi.waitFor(() => { expect(sourceCalls).toBe(1) })
expect(connected).not.toHaveBeenCalled()
})
it('rejects a generation whose source ends during readiness and retries', async () => {
const source = new FakeGenerationSource()
source.holdReady = true
const states: ConnectionState[] = []
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(api, api.generation, {
const controller = new ConnectionController(source.source, {
onConnected: () => { connected++ },
onStateChange: state => states.push(state),
}, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(api.openGenerationCount).toBe(1) })
api.endStreams()
firstDescribe.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, home: '/h', canOpenPath: true }))
await vi.waitFor(() => { expect(describeCalls).toBe(2) })
await vi.waitFor(() => { expect(source.activeCount).toBe(1) })
source.holdReady = false
source.end()
await vi.waitFor(() => { expect(connected).toBe(1) })
expect(states).toEqual(['reconnecting', 'connected'])
} finally {
@@ -181,22 +144,21 @@ describe('connection lifecycle', () => {
{ label: 'ends normally', fail: () => Promise.resolve() },
{
label: 'rejects with a non-Error reason',
// oxlint-disable-next-line typescript/prefer-promise-reject-errors -- non-Error source normalization is the scenario.
fail: () => Promise.reject('fixture offline'),
},
])('retries when the generation source $label before reporting ready', async ({ fail }) => {
const api = new FakeApiClient()
let sourceCalls = 0
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(api, (signal, ready) => {
const source: ConnectionGenerationSource = (signal, ready) => {
sourceCalls++
if (sourceCalls === 1) return fail()
ready({ home: '/h' })
return new Promise<void>((resolve) => {
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
}, { onConnected: () => { connected++ } }, FAST)
}
const controller = new ConnectionController(source, { onConnected: () => { connected++ } }, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(sourceCalls).toBe(2) })
@@ -208,19 +170,19 @@ describe('connection lifecycle', () => {
})
it('rejects and retries a generation whose source never reports ready', async () => {
const api = new FakeApiClient()
api.suppressGenerationReady = true
const source = new FakeGenerationSource()
source.suppressReady = true
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(
api,
api.generation,
source.source,
{ onConnected: () => { connected++ } },
{ ...FAST, generationReadyTimeoutMs: 20 },
)
controller.start()
try {
await vi.waitFor(() => { expect(api.callsOf('host.describe').length).toBeGreaterThan(1) })
await vi.waitFor(() => { expect(source.activeCount).toBeGreaterThan(0) })
await new Promise(resolve => setTimeout(resolve, 45))
expect(connected).toBe(0)
} finally {
controller.stop()
@@ -229,11 +191,11 @@ describe('connection lifecycle', () => {
})
it('emits deduplicated connected/reconnecting state transitions', async () => {
const api = new FakeApiClient()
const source = new FakeGenerationSource()
const states: ConnectionState[] = []
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(api, api.generation, {
const controller = new ConnectionController(source.source, {
onConnected: () => { connected++ },
onStateChange: state => states.push(state),
}, FAST)
@@ -241,7 +203,7 @@ describe('connection lifecycle', () => {
try {
await vi.waitFor(() => { expect(connected).toBe(1) })
expect(states).toEqual(['connected'])
api.failStreams(new Error('torn'))
source.fail(new Error('torn'))
await vi.waitFor(() => { expect(connected).toBe(2) })
expect(states).toEqual(['connected', 'reconnecting', 'connected'])
} finally {
@@ -251,10 +213,10 @@ describe('connection lifecycle', () => {
})
it('does not announce a generation stopped synchronously by its connected state sink', async () => {
const api = new FakeApiClient()
const source = new FakeGenerationSource()
const states: ConnectionState[] = []
let connected = 0
const controller = new ConnectionController(api, api.generation, {
const controller = new ConnectionController(source.source, {
onConnected: () => { connected++ },
onStateChange: (state) => {
states.push(state)
@@ -264,59 +226,58 @@ describe('connection lifecycle', () => {
controller.start()
await vi.waitFor(() => { expect(states).toEqual(['connected']) })
await vi.waitFor(() => { expect(api.openGenerationCount).toBe(0) })
await vi.waitFor(() => { expect(source.activeCount).toBe(0) })
expect(connected).toBe(0)
})
it('deduplicates consecutive reconnecting emissions across two straight failures', async () => {
const api = new FakeApiClient()
const gate = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
let describeCalls = 0
api.onDescribe = () => {
describeCalls++
return describeCalls <= 2 ? Promise.reject(new Error('down')) : gate.promise
}
let sourceCalls = 0
const states: ConnectionState[] = []
let connected = 0
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
const controller = new ConnectionController(api, api.generation, {
const source: ConnectionGenerationSource = (signal, ready) => {
sourceCalls++
if (sourceCalls <= 2) return Promise.reject(new Error('down'))
ready({ home: '/h' })
return new Promise<void>((resolve) => {
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
}
const controller = new ConnectionController(source, {
onConnected: () => { connected++ },
onStateChange: state => states.push(state),
}, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(describeCalls).toBe(3) })
gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0, home: '/h', canOpenPath: true }))
await vi.waitFor(() => { expect(sourceCalls).toBe(3) })
await vi.waitFor(() => { expect(connected).toBe(1) })
expect(states).toEqual(['reconnecting', 'connected']) // two failures, one reconnecting emission
expect(states).toEqual(['reconnecting', 'connected'])
} finally {
controller.stop()
warnSpy.mockRestore()
}
})
it('runs with no sinks at all (every callback slot optional)', async () => {
const api = new FakeApiClient()
const controller = new ConnectionController(api, api.generation, {}, FAST)
it('runs with no sinks at all', async () => {
const source = new FakeGenerationSource()
const controller = new ConnectionController(source.source, {}, FAST)
controller.start()
try {
await vi.waitFor(() => { expect(api.callsOf('host.describe')).toHaveLength(1) })
await new Promise(resolve => setTimeout(resolve, 20))
await vi.waitFor(() => { expect(source.activeCount).toBe(1) })
} finally {
controller.stop()
}
})
it('start() is idempotent (one loop, one stream set)', async () => {
const api = new FakeApiClient()
it('start() is idempotent', async () => {
const source = new FakeGenerationSource()
let connected = 0
const controller = new ConnectionController(api, api.generation, { onConnected: () => { connected++ } }, FAST)
const controller = new ConnectionController(source.source, { onConnected: () => { connected++ } }, FAST)
controller.start()
controller.start()
try {
await vi.waitFor(() => { expect(connected).toBe(1) })
expect(api.openGenerationCount).toBe(1)
expect(api.callsOf('host.describe')).toHaveLength(1)
expect(source.activeCount).toBe(1)
} finally {
controller.stop()
}
packages/client/connection/tests/fake-api.client.ts
-131
View File
@@ -1,131 +0,0 @@
// 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). The generation source is a hand pump.
import type { IApiClient, RpcResponse } from '../src/client/api.ts'
import type { ConnectionGenerationSource } from '../src/client/connection.ts'
import { RpcId } from '../src/client/api.ts'
export interface Deferred<T> {
promise: Promise<T>
resolve(value: T): void
reject(error: unknown): void
}
/** Test-held settlement: the case decides when an RPC lands (history-pending injections etc.). */
export function deferred<T>(): Deferred<T> {
let resolve!: (value: T) => void
let reject!: (error: unknown) => void
const promise = new Promise<T>((res, rej) => {
resolve = res
reject = rej
})
return { promise, resolve, reject }
}
let nextRpc = 0
export function ok<T>(value: T): RpcResponse<T> {
return { rpcId: RpcId(`fake-${nextRpc++}`), result: { ok: true, value } }
}
type StreamItem = { kind: 'end' } | { kind: 'fail'; error: unknown }
interface StreamConn {
feed(item: StreamItem): void
}
export class FakeApiClient implements IApiClient {
/** Chronological call record: [method, payload]. */
readonly calls: { method: string; payload: unknown }[] = []
// Programmable slots (defaults answer OK-empty); reassign per case.
onDescribe: (payload: unknown) => Promise<RpcResponse<{
version: string
cwd: string
attachedSessions: number
home: string
canOpenPath: boolean
}>> =
() => Promise.resolve(ok({
version: '0-fake', cwd: '/f', attachedSessions: 0, home: '/h', canOpenPath: true,
}))
private readonly generationConns: StreamConn[] = []
readonly host: IApiClient['host'] = {
describe: payload => this.record('host.describe', payload, this.onDescribe(payload)),
}
/** When true, the source never reports ready. */
suppressGenerationReady = false
/** When true, ready callbacks remain parked until the test releases them. */
holdGenerationReady = false
private heldOpens: (() => void)[] = []
releaseGenerationReady(): void {
const held = this.heldOpens
this.heldOpens = []
for (const fire of held) fire()
}
readonly generation: ConnectionGenerationSource = (signal, ready) =>
this.openGeneration(signal, ready)
/** End (clean close) or fail (throw) every open stream — reconnect-path material. */
endStreams(): void {
for (const conn of [...this.generationConns]) conn.feed({ kind: 'end' })
}
failStreams(error: unknown): void {
for (const conn of [...this.generationConns]) conn.feed({ kind: 'fail', error })
}
get openGenerationCount(): number {
return this.generationConns.length
}
callsOf(method: string): unknown[] {
return this.calls.filter(c => c.method === method).map(c => c.payload)
}
private record<T>(method: string, payload: unknown, response: Promise<T>): Promise<T> {
this.calls.push({ method, payload })
return response
}
private async openGeneration(
signal: AbortSignal,
onOpen: (host: { readonly home: string }) => void,
): Promise<void> {
const inbox: StreamItem[] = []
let wake: (() => void) | null = null
const conn: StreamConn = {
feed: (item) => {
inbox.push(item)
wake?.()
},
}
this.generationConns.push(conn)
const ready = (): void => { onOpen({ home: '/h' }) }
if (this.holdGenerationReady) this.heldOpens.push(ready)
else if (!this.suppressGenerationReady) ready()
try {
while (!signal.aborted) {
while (inbox.length > 0) {
const item = inbox.shift() as StreamItem
if (item.kind === 'end') return
if (item.kind === 'fail') throw item.error
}
await new Promise<void>((resolve) => {
wake = resolve
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
wake = null
}
} finally {
this.generationConns.splice(this.generationConns.indexOf(conn), 1)
}
}
}
packages/client/connection/tests/fake-generation.client.ts
+80
View File
@@ -0,0 +1,80 @@
/** Test-local programmable Connection generation source. */
import type { ConnectionGenerationSource } from '../src/client/connection.ts'
type StreamItem = { kind: 'end' } | { kind: 'fail'; error: unknown }
interface StreamConnection {
feed(item: StreamItem): void
}
/** Hand-pumped generation source for Connection lifecycle tests. */
export class FakeGenerationSource {
private readonly connections: StreamConnection[] = []
/** When true, the source never reports ready. */
suppressReady = false
/** When true, ready callbacks remain parked until the test releases them. */
holdReady = false
private heldReady: Array<() => void> = []
/** Open one generation. */
readonly source: ConnectionGenerationSource = (signal, ready) => this.open(signal, ready)
/** Release every generation currently parked before readiness. */
releaseReady(): void {
const held = this.heldReady
this.heldReady = []
for (const fire of held) fire()
}
/** End every active generation normally. */
end(): void {
for (const connection of [...this.connections]) connection.feed({ kind: 'end' })
}
/** Fail every active generation. */
fail(error: unknown): void {
for (const connection of [...this.connections]) connection.feed({ kind: 'fail', error })
}
/** Number of currently active generations. */
get activeCount(): number {
return this.connections.length
}
private async open(
signal: AbortSignal,
onReady: (host: { readonly home: string }) => void,
): Promise<void> {
const inbox: StreamItem[] = []
let wake: (() => void) | null = null
const connection: StreamConnection = {
feed: (item) => {
inbox.push(item)
wake?.()
},
}
this.connections.push(connection)
const ready = (): void => { onReady({ home: '/h' }) }
if (this.holdReady) this.heldReady.push(ready)
else if (!this.suppressReady) ready()
try {
while (!signal.aborted) {
while (inbox.length > 0) {
const item = inbox.shift() as StreamItem
if (item.kind === 'end') return
if (item.kind === 'fail') throw item.error
}
await new Promise<void>((resolve) => {
wake = resolve
signal.addEventListener('abort', () => { resolve() }, { once: true })
})
wake = null
}
} finally {
this.connections.splice(this.connections.indexOf(connection), 1)
}
}
}
packages/client/connection/tests/fetch-routes.host.spec.ts
+4 -13
View File
@@ -19,17 +19,16 @@ async function mounted(): Promise<{
}
describe('Connection exact Fetch routes', () => {
it('dispatches owned methods before the transitional fallback', async () => {
it('dispatches owned methods and returns 404 for unclaimed requests', async () => {
const { connection, dispose: disposeFiber } = await mounted()
const route = vi.fn(async (request: Request) =>
Response.json({ query: new URL(request.url).searchParams.get('sessionId') }))
const fallback = vi.fn(async () => new Response('fallback', { status: 418 }))
const dispose = connection.fetch.register({
path: '/api/session.export',
methods: ['GET', 'HEAD'],
fetch: route,
})
const shared = connection.createSharedFetchHandler('/api', { fetch: fallback })
const shared = connection.createSharedFetchHandler('/api')
const response = await shared.fetch(new Request(
'http://host/api/session.export?sessionId=session-1',
@@ -37,16 +36,12 @@ describe('Connection exact Fetch routes', () => {
expect(response.status).toBe(200)
expect(await response.json()).toEqual({ query: 'session-1' })
expect(route).toHaveBeenCalledOnce()
expect(fallback).not.toHaveBeenCalled()
const post = await shared.fetch(new Request('http://host/api/session.export', { method: 'POST' }))
expect(post.status).toBe(418)
expect(fallback).toHaveBeenCalledOnce()
expect(post.status).toBe(404)
await dispose()
const withdrawn = await shared.fetch(new Request('http://host/api/session.export'))
expect(withdrawn.status).toBe(418)
expect(fallback).toHaveBeenCalledTimes(2)
expect(withdrawn.status).toBe(404)
await disposeFiber()
})
@@ -61,10 +56,6 @@ describe('Connection exact Fetch routes', () => {
expect(() => connection.fetch.register({
path: '/api/session.export', methods: ['GET', 'GET'], fetch,
})).toThrow('repeats a method')
expect(() => connection.fetch.register({
path: '/api/session.export', methods: ['POST' as 'GET'], fetch,
})).toThrow('unsupported method')
const dispose = connection.fetch.register({
path: '/api/session.export', methods: ['GET'], fetch,
})
packages/client/connection/tests/fixture-commands.client.spec.ts
+1 -1
View File
@@ -169,7 +169,7 @@ describe('createFixtureApi commands/skills', () => {
})
})
describe('FixtureApiClient command/skill dispatch', () => {
describe('fixture Connection command/skill dispatch', () => {
it('routes the Remote command and skill rows through one state graph', async () => {
const { rpc } = createFixtureFaces()
const commands = await callRemote<{ name: string }[]>(rpc, 'commands/list', { agentId: sid('fx-alpha') })
packages/client/connection/tests/fixture.client.spec.ts
+24 -58
View File
@@ -1,7 +1,5 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import type {
ModelSelection,
RpcMessage,
RpcRequest,
RpcResponse,
RpcResult,
@@ -12,7 +10,7 @@ import { RpcId } from '../src/client/api.ts'
import { decodeStorageRecord } from '@deepseek-ai/dsh-session/chunk-rows'
import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
import {
FixtureApiClient,
createFixtureConnectionRpc,
createFixtureFaces,
type FixtureOptions,
} from '../src/client/fixture.ts'
@@ -21,6 +19,7 @@ import type {
} from '../src/rpc.ts'
import type { DirectoryListing } from '@deepseek-ai/dsh-host-directory-picker/types'
import type { ModelCatalog } from '@deepseek-ai/dsh-api-session-controller/types'
import type { ModelSelection } from '@deepseek-ai/dsh-api-session-controller/types'
const sid = (id: string): SessionId => id as SessionId
type WorkspaceId = string & { readonly __fixtureWorkspaceId: 'WorkspaceId' }
@@ -288,7 +287,7 @@ interface FixtureRemoteEventStream extends AsyncIterable<FixtureRemoteEventFrame
readonly clientId: Promise<string>
}
type FixtureTestApi = ReturnType<typeof createFixtureFaces>['api'] & {
type FixtureTestApi = {
/** The directory-picking Remote namespace as the fixture serves it. */
readonly directoryPickerRemote: {
pick: () => Promise<ConnectionRpcResult<string | null>>
@@ -307,8 +306,8 @@ type FixtureTestApi = ReturnType<typeof createFixtureFaces>['api'] & {
/** Keep existing fixture assertions compact while driving only the new Session Remote endpoints. */
function createFixtureApi(options: FixtureOptions = {}): FixtureTestApi {
const { api, rpc } = createFixtureFaces(options)
return Object.assign(api, {
const { rpc } = createFixtureFaces(options)
return {
directoryPickerRemote: {
pick: () => rpc.call('/api', 'directoryPicker/pick', { args: {} }) as
Promise<ConnectionRpcResult<string | null>>,
@@ -327,7 +326,7 @@ function createFixtureApi(options: FixtureOptions = {}): FixtureTestApi {
remoteEvents: (signal: AbortSignal) => openFixtureRemoteEvents(rpc, signal),
answerRemoteEvent: (result: FixtureRemoteEventResult) =>
rpc.call('/api', '$events/result', { args: result }),
})
}
}
/** The fixture's Credentials Remote endpoints over the shared RPC carrier. */
@@ -1103,16 +1102,6 @@ describe('createFixtureApi', () => {
expect(remaining.map(frame => frame.event)).toEqual(['user-questions/request'])
})
it('describe answers the fixture identity', async () => {
const api = createFixtureApi()
const response = await api.host.describe(req({}))
expect(response.result).toMatchObject({
ok: true, value: { version: '0.0.0-fixture', attachedSessions: 1, home: '/home/fixture' },
})
const empty = await createFixtureApi({ empty: true }).host.describe(req({}))
expect(empty.result).toMatchObject({ ok: true, value: { attachedSessions: 0 } })
})
it('createDirectory under the root mints /name whose listing and crumbs share the identity', async () => {
const api = createFixtureApi()
const created = await api.directoryPickerRemote.createDirectory('/', 'srv')
@@ -1613,38 +1602,16 @@ describe('createFixtureApi', () => {
})
})
describe('FixtureApiClient (protocol-level fake carrier)', () => {
describe('fixture Connection RPC', () => {
afterEach(() => {
vi.restoreAllMocks()
vi.unstubAllGlobals()
})
it('doFetch is an unreachable tripwire (all protocol paths overridden)', () => {
const client = new FixtureApiClient()
// Protected at compile time only; reach it directly to pin the tripwire message.
expect(() => (client as unknown as { doFetch(): Promise<Response> }).doFetch()).toThrow(/doFetch must be unreachable/)
})
it('mints request ids and taps unary request/response envelopes without touching doFetch', async () => {
const client = new FixtureApiClient()
const tapped: RpcMessage[] = []
client.subscribeEnvelopes(batch => tapped.push(...batch))
const response = await client.host.describe({})
expect(response.result.ok).toBe(true)
await vi.waitFor(() => {
const kinds = tapped.map(m => m.type)
expect(kinds).toContain('client-request')
expect(kinds).toContain('server-response')
})
const request = tapped.find(m => m.type === 'client-request')
const reply = tapped.find(m => m.type === 'server-response')
expect(request?.rpcId).toBe(reply?.rpcId) // echo discipline holds through the fake carrier
})
it('covers the whole unary dispatch table', async () => {
const client = new FixtureApiClient()
const sessions = createSessionClient(client.rpc)
const workspaces = createWorkspaceClient(client.rpc)
it('covers the migrated Remote dispatch table', async () => {
const rpc = createFixtureConnectionRpc()
const sessions = createSessionClient(rpc)
const workspaces = createWorkspaceClient(rpc)
expect((await sessions.search(
{ query: 'fixture' },
new AbortController().signal,
@@ -1655,8 +1622,7 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
expect((await sessions.history({ sessionId: id })).result.ok).toBe(true)
expect((await sessions.prompt({ sessionId: id, mode: 'queue', content: [{ type: 'text', text: '嗨' }] })).result.ok).toBe(true)
expect((await sessions.cancel({ sessionId: id })).result.ok).toBe(true)
expect((await client.host.describe({})).result.ok).toBe(true)
expect((await readWorkspaceBaseline(createWorkspaceRemote(client.rpc))).items).not.toHaveLength(0)
expect((await readWorkspaceBaseline(createWorkspaceRemote(rpc))).items).not.toHaveLength(0)
const workspace = await workspaces.create({ path: '/tmp/fixture-workspaces/via-client' })
if (!workspace.result.ok) throw new Error('workspace create failed')
expect(workspace.result.value.workspace.title).toBe('via-client')
@@ -1672,13 +1638,13 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
})
it('folds the goal lifecycle over the Goal Remotes', async () => {
const client = new FixtureApiClient()
const sessions = createSessionClient(client.rpc)
const rpc = createFixtureConnectionRpc()
const sessions = createSessionClient(rpc)
const created = await sessions.create({})
if (!created.result.ok) throw new Error('create failed')
const id = created.result.value.sessionId
const goal = (endpoint: string, args: Record<string, unknown>) =>
client.rpc.call('/api', endpoint, { args: { agentId: id, ...args } })
rpc.call('/api', endpoint, { args: { agentId: id, ...args } })
// create → edit → pause → resume → complete → clear; each mutation advances the CAS
// revision by one (state rides the projection frames).
@@ -1717,17 +1683,17 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
vi.stubGlobal('location', {
search: '?fixture=empty&fixturePrompt=reject&fixtureFrames=workspace-first',
})
const client = new FixtureApiClient()
const sessions = createSessionClient(client.rpc)
const workspaces = createWorkspaceClient(client.rpc)
const workspaceRemote = createWorkspaceRemote(client.rpc)
const rpc = createFixtureConnectionRpc()
const sessions = createSessionClient(rpc)
const workspaces = createWorkspaceClient(rpc)
const workspaceRemote = createWorkspaceRemote(rpc)
await expect(sessions.list({})).resolves.toMatchObject({ result: { ok: true, value: { items: [] } } })
const made = await workspaces.create({ path: '/tmp/fixture-workspaces/query-workspace' })
if (!made.result.ok) throw new Error('workspace create failed')
const hostAbort = new AbortController()
const workspaceAbort = new AbortController()
const hostFrames = collectValues(
openFixtureRemoteEvents(client.rpc, hostAbort.signal),
openFixtureRemoteEvents(rpc, hostAbort.signal),
hostAbort,
frames => frames.length === 1,
)
@@ -1761,8 +1727,8 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
it('maps attach-failure and dropped-response query scenarios', async () => {
vi.stubGlobal('location', { search: '?fixture&fixtureAttach=fail' })
const partial = new FixtureApiClient()
const partialResult = await createSessionClient(partial.rpc).create({
const partial = createFixtureConnectionRpc()
const partialResult = await createSessionClient(partial).create({
workspaceId: 'fx-ws-fixture' as WorkspaceId,
sessionId: sid('fx-query-partial'),
})
@@ -1772,8 +1738,8 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
})
vi.stubGlobal('location', { search: '?fixture&fixtureSessionCreate=drop-response' })
const dropped = new FixtureApiClient()
await expect(createSessionClient(dropped.rpc).create({
const dropped = createFixtureConnectionRpc()
await expect(createSessionClient(dropped).create({
workspaceId: 'fx-ws-fixture' as WorkspaceId,
sessionId: sid('fx-query-dropped'),
})).rejects.toThrow(/dropped session\.create response/)
packages/client/connection/tests/node-half.host.spec.ts
+3 -9
View File
@@ -6,11 +6,9 @@ import { Context } from '@deepseek-ai/cordis'
import { describe, expect, it } from 'vitest'
import type { AddressInfo } from 'node:net'
import type { IncomingMessage, ServerResponse } from 'node:http'
import type { ApiProxy } from '@deepseek-ai/dsh-host-apiproxy/api'
import type { AttachmentStore } from '@deepseek-ai/dsh-attachment'
import { RpcId, type ClientRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
import type { WebServer, WebRoute, WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver'
import { API_PATH, apply, inject, type HostConnectionHandle } from '../src/index.ts'
import { API_PATH, RpcId, apply, inject, type ClientRequest, type HostConnectionHandle } from '../src/index.ts'
import { DEFAULT_MAX_REQUEST_BODY_BYTES } from '../src/http-bridge.ts'
import { provideBrowserCredentials } from './browser-credentials.ts'
@@ -94,7 +92,6 @@ async function mounted(config?: { trustedHosts?: string[] }): Promise<{
const upgrades: WebUpgradeRoute[] = []
provideBrowserCredentials(ctx)
ctx.provide('webServer', fakeHttpServer(routes, upgrades) as WebServer)
ctx.provide('apiProxy', {} as unknown as ApiProxy)
const fiber = ctx.plugin({ inject: [...inject], apply }, config)
await fiber.await()
return {
@@ -131,7 +128,6 @@ describe('connection node half', () => {
ctx.provide('attachments', {
imageLimits: { maxMessageImageBytes: 20 * 1024 * 1024 },
} as AttachmentStore)
ctx.provide('apiProxy', {} as ApiProxy)
await expect(apply(ctx, { maxRequestBodyBytes: 1024 }))
.rejects.toThrow(/must be at least .* aggregate image limit/)
expect(routes).toHaveLength(0)
@@ -143,7 +139,6 @@ describe('connection node half', () => {
const ctx = new Context()
provideBrowserCredentials(ctx)
ctx.provide('webServer', fakeHttpServer(routes, upgrades) as WebServer)
ctx.provide('apiProxy', {} as unknown as ApiProxy)
const fiber = ctx.plugin({ inject: [...inject], apply }, { trustedHosts: ['harness.internal/path'] })
await expect(fiber).rejects.toThrow(/not a bare host\[:port\] authority/)
expect(routes).toHaveLength(0)
@@ -243,7 +238,7 @@ describe('connection node half', () => {
await dispose()
})
it('provides a disposable dedicated RPC channel without requiring apiProxy', async () => {
it('provides a disposable dedicated RPC channel', async () => {
const ctx = new Context()
const routes: WebRoute[] = []
provideBrowserCredentials(ctx)
@@ -292,12 +287,11 @@ describe('connection node half', () => {
expect(routes).toHaveLength(0)
})
it('dispatches claimed /api endpoints before the API Proxy fallback and withdraws the claim', async () => {
it('dispatches claimed /api endpoints and withdraws the claim', async () => {
const ctx = new Context()
const routes: WebRoute[] = []
provideBrowserCredentials(ctx)
ctx.provide('webServer', fakeHttpServer(routes, []) as WebServer)
ctx.provide('apiProxy', {} as unknown as ApiProxy)
const fiber = ctx.plugin({ inject: [...inject], apply }, { trustedHosts: ['harness.example'] })
await fiber.await()
const connection = ctx.get('connection') as HostConnectionHandle
packages/client/connection/tests/rpc-schema.host.spec.ts
+58
View File
@@ -0,0 +1,58 @@
import { describe, expect, it } from 'vitest'
import { RpcId, transportError } from '../src/rpc.ts'
import {
clientRequestSchema,
rpcErrorSchema,
rpcIdSchema,
rpcMessageSchema,
rpcResultSchema,
serverResponseSchema,
} from '../src/rpc-schema.ts'
import { z } from 'zod'
describe('Connection RPC schema', () => {
it('brands any validated string correlation id', () => {
expect(RpcId('abc')).toBe('abc')
expect(rpcIdSchema.parse('')).toBe('')
expect(() => rpcIdSchema.parse(42)).toThrow()
})
it('folds transport exceptions into an internal failure', () => {
expect(transportError(new Error('wire down'))).toEqual({
ok: false,
error: { code: 'internal', message: 'wire down', details: {} },
})
expect(transportError('raw')).toMatchObject({
ok: false,
error: { code: 'internal', message: 'raw' },
})
})
it('validates generic failures and both result branches', () => {
expect(rpcErrorSchema.parse({ code: 'domain-failure', message: 'failed', details: { id: 'x' } }))
.toEqual({ code: 'domain-failure', message: 'failed', details: { id: 'x' } })
expect(() => rpcErrorSchema.parse({ code: 1, message: 'failed', details: {} })).toThrow()
expect(() => rpcErrorSchema.parse({ code: 'failed', message: 'failed', details: [] })).toThrow()
const schema = rpcResultSchema(z.object({ n: z.number() }))
expect(schema.parse({ ok: true, value: { n: 1 } })).toEqual({ ok: true, value: { n: 1 } })
expect(schema.parse({ ok: false, error: { code: 'failed', message: 'x', details: {} } }))
.toMatchObject({ ok: false })
expect(() => schema.parse({ ok: true, error: {} })).toThrow()
})
it('validates both envelope directions and valueless success', () => {
const request = { type: 'client-request', rpcId: 'r1', method: 'settings/describe', payload: { args: {} } }
const response = { type: 'server-response', rpcId: 'r1', result: { ok: true, value: 1 } }
expect(clientRequestSchema.parse(request).method).toBe('settings/describe')
expect(serverResponseSchema.parse(response).rpcId).toBe('r1')
for (const message of [request, response]) expect(rpcMessageSchema.parse(message)).toBeTruthy()
expect(() => rpcMessageSchema.parse({ type: 'other', rpcId: 'x' })).toThrow()
expect(() => clientRequestSchema.parse({ type: 'client-request', rpcId: 'r1' })).toThrow()
expect(() => serverResponseSchema.parse({ type: 'server-response', rpcId: 'r1' })).toThrow()
expect(() => serverResponseSchema.parse({ type: 'server-response', rpcId: 'r1', result: {} })).toThrow()
expect(serverResponseSchema.parse({
type: 'server-response', rpcId: 'r1', result: { ok: true },
}).rpcId).toBe('r1')
})
})
packages/client/connection/tsconfig.client.json
-4
View File
@@ -13,7 +13,6 @@
"src/client/index.ts",
"src/client/random-uuid.ts",
"src/client/rpc.ts",
"src/client/web-api-client.ts",
"src/loopback-hostname.ts",
"src/rpc.ts"
],
@@ -39,9 +38,6 @@
{
"path": "../../settings/settings"
},
{
"path": "../../host/apiproxy"
},
{
"path": "../../host/directory-picker"
},
packages/client/connection/tsconfig.host.json
+5 -1
View File
@@ -14,6 +14,7 @@
"src/invariant.ts",
"src/loopback-hostname.ts",
"src/rpc-host.ts",
"src/rpc-schema.ts",
"src/rpc.ts"
],
"references": [
@@ -24,13 +25,16 @@
"path": "../../credentials/credentials"
},
{
"path": "../../host/apiproxy"
"path": "../../core/session"
},
{
"path": "../../host/webserver"
},
{
"path": "../../runtime-diagnostics/invariants"
},
{
"path": "../../util/brand"
}
]
}
packages/client/tsdown.client.ts
+1 -1
View File
@@ -58,7 +58,7 @@ function styleInjectionModule(
* Everything else under @deepseek-ai/* is either a module-table entry
* (external) or a leak the purity gate rejects.
*/
export const INLINE_SAFE = /^(?:@deepseek-ai\/dsh-(?:host-apiproxy|file-reference|session|llm|tools|brand|util-crypto|util-workspace-path)(?:\/|$)|@deepseek-ai\/dsh-token-meter\/client$)/
export const INLINE_SAFE = /^(?:@deepseek-ai\/dsh-(?:file-reference|session|llm|tools|brand|util-crypto|util-workspace-path)(?:\/|$)|@deepseek-ai\/dsh-token-meter\/client$)/
/**
* Vendored framework libraries: rescoped into @deepseek-ai, so the gate below
packages/credentials/authorization/src/types.ts
+1 -1
View File
@@ -1,6 +1,6 @@
/**
* Wire-safe authorization types, free of cordis/service imports so browser type
* chains (apiproxy api → client) can consume them without loading this
* chains can consume them without loading this
* package's Context augmentation.
* @module @deepseek-ai/dsh-authorization/types
*/
packages/experimental/webworker-packer/src/rules.ts
-1
View File
@@ -65,7 +65,6 @@ export const PAGE_ASSETS: readonly string[] = [
export const IMAGE_ENTRY_SEEDS: readonly string[] = [
'@deepseek-ai/dsh-app-boot',
'@deepseek-ai/dsh-cmdline',
'@deepseek-ai/dsh-host-apiproxy',
'@deepseek-ai/cordis',
'@deepseek-ai/cordis-plugin-include',
'js-yaml',
packages/experimental/webworker-packer/tsconfig.json
+3
View File
@@ -11,6 +11,9 @@
"src"
],
"references": [
{
"path": "../../../vendor/include"
},
{
"path": "../webworker-runtime"
},
packages/experimental/webworker-runtime/package.json
+2 -2
View File
@@ -41,8 +41,8 @@
"peerDependencies": {
"@deepseek-ai/cordis": "workspace:^",
"@deepseek-ai/cordis-plugin-loader": "workspace:^",
"@deepseek-ai/dsh-client-connection": "workspace:^",
"@deepseek-ai/dsh-client-modules": "workspace:^",
"@deepseek-ai/dsh-host-apiproxy": "workspace:^",
"@deepseek-ai/dsh-host-webserver": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^"
},
@@ -51,8 +51,8 @@
"@deepseek-ai/cordis-plugin-loader": "workspace:^",
"@deepseek-ai/dsh-api-gateway": "workspace:^",
"@deepseek-ai/dsh-bash-sandbox": "workspace:^",
"@deepseek-ai/dsh-client-connection": "workspace:^",
"@deepseek-ai/dsh-client-modules": "workspace:^",
"@deepseek-ai/dsh-host-apiproxy": "workspace:^",
"@deepseek-ai/dsh-host-webserver": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-sandbox-local": "workspace:^",
packages/experimental/webworker-runtime/src/client/api-client.ts
-31
View File
@@ -1,31 +0,0 @@
/**
* Page-side unary API carrier over the postMessage tunnel. Gateway Remote
* streams use the tunnel's dedicated logical-stream frames instead of this
* fetch-shaped API path.
*/
import { AbstractApiClient } from '@deepseek-ai/dsh-host-apiproxy/client'
import type { WorkerTunnel } from './client.ts'
/** API client whose requests travel the worker tunnel instead of the network. */
export class WorkerApiClient extends AbstractApiClient {
private readonly tunnel: WorkerTunnel
/**
* Bind the carrier to a tunnel.
* @param tunnel - page half of the worker tunnel.
*/
constructor(tunnel: WorkerTunnel) {
super()
this.tunnel = tunnel
}
/**
* Send one request through the tunnel.
* @param input - request URL.
* @param init - fetch init; the tunnel honours method, headers, body, and signal.
* @returns the reconstructed response.
*/
protected doFetch(input: URL, init?: RequestInit): Promise<Response> {
return this.tunnel.fetch(input, init)
}
}
packages/experimental/webworker-runtime/src/client/index.ts
-4
View File
@@ -10,12 +10,10 @@
*/
import { IMAGE_FILE_NAME } from '../image-layout.ts'
import { PREVIEW_FIXTURE_MANIFEST_FILE } from '../fixture-manifest.ts'
import { WorkerApiClient } from './api-client.ts'
import { WorkerTunnel, type TunnelFetch } from './client.ts'
import { applyIndexInjections } from './apply-injections.ts'
import { choosePreviewSource } from './source-chooser.ts'
export { WorkerApiClient } from './api-client.ts'
export { WorkerTunnel, type TunnelFetch } from './client.ts'
export { applyIndexInjections } from './apply-injections.ts'
export { IMAGE_FILE_NAME } from '../image-layout.ts'
@@ -27,7 +25,6 @@ export {
/** Transport global the connection plugin reads instead of building an HTTP carrier. */
interface ClientTransportGlobal {
__DSH_TRANSPORT__?: {
createApiClient: () => WorkerApiClient
fetch: TunnelFetch
openStream: (endpoint: string, payload: unknown, signal: AbortSignal) => AsyncIterable<unknown>
loadBundle: (url: string) => Promise<void>
@@ -147,7 +144,6 @@ export async function connectWorkerHost(worker: Worker, options?: WorkerHostConn
)
const payload = await tunnel.bootPayload()
;(globalThis as ClientTransportGlobal).__DSH_TRANSPORT__ = {
createApiClient: () => new WorkerApiClient(tunnel),
fetch: (input, init) => tunnel.fetch(input, init),
openStream: (endpoint, payload, signal) => tunnel.open(endpoint, payload, signal),
loadBundle: (url: string) => tunnel.loadBundle(url),
packages/experimental/webworker-runtime/src/node/builtins.ts
+1 -1
View File
@@ -2,7 +2,7 @@
* The Node-compatibility table, in one place. Two consumers share it, and they
* must resolve to the same module instances:
* - the worker vite build aliases these specifiers for code bundled statically
* into the worker (vendored loader, apiproxy, …);
* into the worker (vendored loader, Connection, …);
* - the worker module loader answers `require('node:fs')` from VFS-loaded
* modules out of this table, before bare-name resolution.
* Anything absent here fails loudly at resolution instead of resolving to an
packages/experimental/webworker-runtime/src/node/external_packages/ws.ts
+1 -1
View File
@@ -1,6 +1,6 @@
/**
* `ws` stub. `WebSocketDownlinks` constructs a `WebSocketServer` in a field
* initializer as soon as apiProxy is present, so the class must be constructible;
* initializer as soon as Connection is present, so the class must be constructible;
* no method is ever reached because the fake HTTP server never emits `upgrade`
* (the tunnel carries downstream events over the SSE branch instead).
*/
packages/experimental/webworker-runtime/src/worker-host.ts
+5 -35
View File
@@ -23,6 +23,7 @@
*/
import { setActiveModuleLoader, WorkerModuleLoader, type StaticModuleFactory } from './module-system/module-loader.ts'
import type { TypertGateway } from '@deepseek-ai/dsh-api-gateway'
import type { HostConnectionHandle } from '@deepseek-ai/dsh-client-connection'
import type { AlsCausality } from './polyfill/async-context/als-runtime.ts'
import { dirname, join } from './module-system/posix-path.ts'
import { installProcessGlobal } from './node/globals/process.ts'
@@ -242,19 +243,15 @@ export function createWorkerHost(options: WorkerHostOptions): WorkerHost {
})
context = ctx
const apiProxy = ctx.get('apiProxy')
if (apiProxy === undefined) throw new Error('webworker host: the tree activated without an apiProxy service')
const connection = ctx.get('connection') as HostConnectionHandle | undefined
if (connection === undefined) throw new Error('webworker host: the tree activated without a Connection service')
const typertGateway = ctx.get('typertGateway') as TypertGateway | undefined
if (typertGateway === undefined) {
throw new Error('webworker host: the tree activated without a typertGateway service')
}
const { toFetchHandler } = require('@deepseek-ai/dsh-host-apiproxy') as {
toFetchHandler: (api: unknown) => { fetch(request: Request): Promise<Response> }
}
const shared = ctx.get('connection') !== undefined
const handler = directFetchHandler(ctx, toFetchHandler(apiProxy))
const handler = connection.createSharedFetchHandler('/api')
const usage = loader.usage()
console.info(`webworker host: tree active (modules=${String(usage.modules)}, data overlays=${String(overlays.length)}, preset root overlay=${presetOverlay ? 'applied' : 'already in roster'}, direct lane=${shared ? 'connection.createSharedFetchHandler (interceptors kept)' : 'api surface only'}, als causality=${options.alsCausality === undefined ? 'inert' : 'snapshot/restore'}, image lowering=${LOWERING_VERSION})`)
console.info(`webworker host: tree active (modules=${String(usage.modules)}, data overlays=${String(overlays.length)}, preset root overlay=${presetOverlay ? 'applied' : 'already in roster'}, direct lane=connection.createSharedFetchHandler, als causality=${options.alsCausality === undefined ? 'inert' : 'snapshot/restore'}, image lowering=${LOWERING_VERSION})`)
tunnel.serve({
directFetch: (request: Request) => handler.fetch(request),
@@ -349,33 +346,6 @@ function requireLoweredImage(vfs: MemoryVfs, path: string): void {
}
}
/**
* Build the tunnel's direct API entry.
*
* The core API surface alone is not the whole `/api` channel: Typert RPC
* endpoints (`/api/<service>/<method>`) are served by an interceptor the gateway
* registers on the Connection service, and answer 404 from the core routes. The
* Connection service composes both halves in `createSharedFetchHandler`, whose
* fallback — not the composition — carries network authentication and trust, so
* composing it here keeps every interceptor while leaving out the fences the
* worker-local direct lane exists to bypass.
* @param ctx - Booted host context.
* @param core - Fetch handler over the API surface.
* @returns Handler covering interceptors and the core surface.
*/
function directFetchHandler(
ctx: HostContext,
core: { fetch(request: Request): Promise<Response> },
): { fetch(request: Request): Promise<Response> } {
const connection = ctx.get('connection') as {
createSharedFetchHandler(
channel: '/api',
fallback: { fetch(request: Request): Promise<Response> },
): { fetch(request: Request): Promise<Response> }
} | undefined
return connection?.createSharedFetchHandler('/api', core) ?? core
}
/**
* The shipped preset root, as the application layer that owns the composition
* supplies it.
packages/experimental/webworker-runtime/tsconfig.json
+1 -1
View File
@@ -27,7 +27,7 @@
"path": "../../client/modules"
},
{
"path": "../../host/apiproxy"
"path": "../../client/connection/tsconfig.host.json"
},
{
"path": "../../host/webserver"
packages/extensions/cordis-client-runner/src/client/api-catalog.ts
+16 -12
View File
@@ -485,13 +485,25 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'ConnectionConfig',
declaration: 'export interface ConnectionConfig {\n backoffBaseMs?: number;\n backoffFactor?: number;\n backoffMaxMs?: number;\n generationReadyTimeoutMs?: number;\n}',
},
{
name: 'ConnectionGeneration',
declaration: 'export interface ConnectionGeneration {\n readonly id: number;\n readonly host: ConnectionHostInfo;\n}',
},
{
name: 'ConnectionGenerationSource',
declaration: 'export type ConnectionGenerationSource = (signal: AbortSignal, ready: () => void) => Promise<void>;',
declaration: 'export type ConnectionGenerationSource = (signal: AbortSignal, ready: (host: ConnectionHostInfo) => void) => Promise<void>;',
},
{
name: 'ConnectionGenerationState',
declaration: 'export interface ConnectionGenerationState {\n getSnapshot(): ConnectionGeneration | undefined;\n subscribe(listener: () => void): () => void;\n}',
},
{
name: 'ConnectionHandle',
declaration: 'export interface ConnectionHandle {\n readonly api: IApiClient;\n readonly isLoopback: boolean;\n readonly hostDescription: HostDescriptionSource;\n readonly rpc: ClientConnectionRpc;\n registerGenerationSource(source: ConnectionGenerationSource): () => void;\n start(sinks: ConnectionSinks, config?: ConnectionConfig): {\n stop(): void;\n };\n}',
declaration: 'export interface ConnectionHandle {\n readonly isLoopback: boolean;\n readonly generation: ConnectionGenerationState;\n readonly rpc: ClientConnectionRpc;\n registerGenerationSource(source: ConnectionGenerationSource): () => void;\n start(sinks: ConnectionSinks, config?: ConnectionConfig): {\n stop(): void;\n };\n}',
},
{
name: 'ConnectionHostInfo',
declaration: 'export interface ConnectionHostInfo {\n readonly home: string;\n}',
},
{
name: 'ConnectionRpcFailure',
@@ -503,7 +515,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'ConnectionSinks',
declaration: 'export interface ConnectionSinks {\n onConnected?: (description: HostDescription) => void;\n onStateChange?: (state: ConnectionState) => void;\n}',
declaration: 'export interface ConnectionSinks {\n onConnected?: (host: ConnectionHostInfo) => void;\n onStateChange?: (state: ConnectionState) => void;\n}',
},
{
name: 'ConnectionState',
@@ -525,14 +537,6 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'HooksSources',
declaration: 'export type HooksSources = Record<string, HostObservable<unknown>>;',
},
{
name: 'HostDescription',
declaration: 'export type HostDescription = import(\'@deepseek-ai/dsh-host-apiproxy/api\').ResponseValue<\'host.describe\'>;',
},
{
name: 'HostDescriptionSource',
declaration: 'export interface HostDescriptionSource {\n getSnapshot(): HostDescription | undefined;\n subscribe(listener: () => void): () => void;\n}',
},
{
name: 'HostObservable',
declaration: 'export type HostObservable<T> = ObservableSnapshot<T>;',
@@ -651,7 +655,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
},
{
name: 'RemoteStream',
declaration: 'export class RemoteStream<Item> implements AsyncIterable<RemoteStreamItem<Item>> {\n constructor(private readonly connection: Pick<ConnectionHandle, \'hostDescription\'>, private readonly options: RemoteStreamOptions<Item>);\n get signal(): AbortSignal;\n restart(): void;\n dispose(): Promise<void>;\n [Symbol.asyncIterator](): AsyncIterator<RemoteStreamItem<Item>>;\n}',
declaration: 'export class RemoteStream<Item> implements AsyncIterable<RemoteStreamItem<Item>> {\n constructor(private readonly connection: Pick<ConnectionHandle, \'generation\'>, private readonly options: RemoteStreamOptions<Item>);\n get signal(): AbortSignal;\n restart(): void;\n dispose(): Promise<void>;\n [Symbol.asyncIterator](): AsyncIterator<RemoteStreamItem<Item>>;\n}',
},
{
name: 'RemoteStreamCarrierError',
packages/extensions/cordis-client-runner/src/client/slot-catalog.ts
+1 -3
View File
@@ -2133,9 +2133,7 @@ export const CLIENT_SLOT_API: readonly ClientSlotEntry[] = [
ownerProps: [
'/** Standard owner currency supplied to every atomic Tool view. */\nexport interface ToolCallOwnerProps {\n /** Tool call identity, stable across running and settled forms. */\n callId: string\n /** Wire Tool name and keyed dispatch value. */\n toolName: string\n /** Frozen running call or settled result node. */\n block: ToolCallBlock\n /** Session workspace root for relative summaries. */\n cwd?: string | undefined\n /** Host account home; POSIX home-rooted summaries display as `~`. */\n home?: string | undefined\n /** Open a Tool argument path through the Host. */\n openFile: (path: string) => void\n /** Inspect this call in the trajectory view when available. */\n inspect?: (() => void) | undefined\n}',
],
ownerPropsReferences: [
'Wire',
],
ownerPropsReferences: [],
standardProps: [
'useWorkspaces: SnapshotSelectorHook<WorkspaceSnapshot>',
'useSessions: UseSessions',
packages/interaction/user-approval/src/types.ts
+1 -1
View File
@@ -1,6 +1,6 @@
/**
* Wire-safe approval identifiers and outcome vocabulary, free of
* cordis/service imports so browser type chains (apiproxy api → client) can
* cordis/service imports so browser type chains can
* consume them without loading this package's Context augmentation.
* @module @deepseek-ai/dsh-user-approval/types
*/
scripts/client-bundle-purity.spec.ts
+3 -4
View File
@@ -92,7 +92,6 @@ describe('client bundle purity gate', () => {
})
it('lets inline-safe wire layers inline', () => {
expect(resolveId('@deepseek-ai/dsh-host-apiproxy/api')).toBeNull()
expect(resolveId('@deepseek-ai/dsh-session/surface')).toBeNull()
expect(resolveId('@deepseek-ai/dsh-brand')).toBeNull()
expect(resolveId('@deepseek-ai/dsh-token-meter/client')).toBeNull()
@@ -218,10 +217,10 @@ describe('client bundle debug artifacts', () => {
if (transform === undefined) throw new Error('client sourcemap path transform missing')
const sourceMapPath = clientSourceMapPath('client/connection')
const workspaceSource = transform('../../../host/apiproxy/src/api/rpc.ts', sourceMapPath)
expect(workspaceSource).toBe('../../../packages/host/apiproxy/src/api/rpc.ts')
const workspaceSource = transform('../src/rpc.ts', sourceMapPath)
expect(workspaceSource).toBe('../../../packages/client/connection/src/rpc.ts')
const resolved = new URL(workspaceSource, 'https://dsh.test/plugins/@deepseek-ai/dsh-client-connection/client.js.map')
expect(resolved.pathname).toBe('/packages/host/apiproxy/src/api/rpc.ts')
expect(resolved.pathname).toBe('/packages/client/connection/src/rpc.ts')
const dependencySource = '../../../../node_modules/.pnpm/zod@4.4.3/node_modules/zod/index.js'
expect(transform(dependencySource, sourceMapPath)).toBe(dependencySource)