Merge branch 'stack/agent-profiles-3-wire' into stack/agent-profiles-5-web-ui

The Client API carrier's `agentPresets` member was the one member of its class
without an `IApiClient[...]` annotation. Inferring it inlined `AgentPresetEntry`
into the emitted declaration by the specifier TS picks — the host `index.ts` —
dragging the whole gateway, and with it the host `Context` merges, into every
Client program importing the carrier. Annotated like its siblings.

`ApiRemoteAgentOptions.setup` now takes the inspected session rather than its
header alone: this layer resolves a resumed session's preset from the LOG,
because a session that switched while blank ran its turns under the newer
composition and the header is written once at creation.

Conflicts:
	apps/web/tests/snapshots/*/*.expected.md
	packages/client/ui-conversation/src/client/skeleton/InputBar.tsx
	packages/host/apiproxy/src/api-proxy.ts
	scripts/doc-budgets.manifest.json
This commit is contained in:
Yichen Jiang
2026-08-08 15:00:31 +08:00
649 changed files with 21091 additions and 2838 deletions

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/client/connection/README.md
README.md: 1393e79aacecbbf7b186f19e4c42269595854b0e
README.zh.md: 70380ceba1b16b2970e947fb6cd9b2af9085ae51
README.md: 161e34c4b6018625fb690e178eb9a9f8ac0ef21b
README.zh.md: d17012cc89c02a1b11f16d126b7c0cafe67fb2a0

View File

@@ -2,7 +2,7 @@
English | [中文](README.zh.md)
Wire consumer layer: the client plugin's apply mounts `ctx.connection` (shared api client + current-page loopback state + single-consumer stream-loop starter); the export face carries the wire contract types, the `AbstractApiClient` seam, and the loop's sink/config types. The browser carrier uses HTTP POST for unary and respond operations and opens one downlink-only WebSocket each for `events.mux` and `events.host`; the in-process carrier satisfies the same two-stream abstraction. Loopback hostname classification stays package-internal: the `/api` Host fence and WebSocket upgrades use it directly, while other client plugins consume the derived `ctx.connection.isLoopback` state. The node half's `/api` route pins the privileged method set (`host.pickDirectory`, `host.openPath`, and the whole configuration plane — `settings.describe`/`openDocument`/`update`/`replace`/`mutate` and `credentials.describe`/`set`/`unset`; reads and native actions included, since describing returns the exposed configuration, opening acts on the Host desktop, and probing an arbitrary reference reports where a credential comes from) to loopback by passing the trust fence with an empty trust list — a declared `trustedHosts` authority reaches every other method, while these stay loopback-local until a real authentication layer exists. The platform carriers and ConnectionController loop are package-internal; apply selects and drives them. The downlink boundary is documented in the [WebSocket downlink carrier Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md); the protocol contract is api-contracts v3 §3.
Wire consumer layer: the client plugin's apply mounts `ctx.connection` (shared api client + current-page loopback state + single-consumer stream-loop starter); the export face carries the wire contract types, the `AbstractApiClient` seam, and the loop's sink/config types. The browser carrier uses HTTP POST for unary and respond operations and opens one downlink-only WebSocket each for `events.mux` and `events.host`; the in-process carrier satisfies the same two-stream abstraction. The Host half owns the single `/api` route and its Fetch bridge; a registered TypeRT interceptor claims its Remote endpoints before the API Proxy fallback. Loopback hostname classification stays package-internal: the `/api` Host fence and WebSocket upgrades use it directly, while other client plugins consume the derived `ctx.connection.isLoopback` state. The node half's `/api` route pins the privileged method set (`host.pickDirectory`, `host.openPath`, and the whole configuration plane — `settings.describe`/`openDocument`/`update`/`replace`/`mutate` and `credentials.describe`/`set`/`unset`; reads and native actions included, since describing returns the exposed configuration, opening acts on the Host desktop, and probing an arbitrary reference reports where a credential comes from) to loopback by passing the trust fence with an empty trust list — a declared `trustedHosts` authority reaches every other method, while these stay loopback-local until a real authentication layer exists. The platform carriers and ConnectionController loop are package-internal; apply selects and drives them. The downlink boundary is documented in the [WebSocket downlink carrier Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md); the protocol contract is api-contracts v3 §3.
## /api browser-trust fence

View File

@@ -2,7 +2,7 @@
[English](README.md) | 中文
协议消费层:客户端插件的 apply 会挂载 `ctx.connection`(共享 API 客户端 + 当前页面的 loopback 状态 + 单消费方流循环启动器);导出表层携带协议契约类型、`AbstractApiClient` seam以及循环的 sink配置类型。浏览器载体以 HTTP POST 发送 unaryrespond并为 `events.mux``events.host` 各开一条只下行的 WebSocket进程内载体满足同一双流抽象。Loopback hostname 判定逻辑留在包内部:`/api` Host fence 与 WebSocket upgrade 会直接使用它,其他客户端插件则消费派生的 `ctx.connection.isLoopback` 状态。node 半侧的 `/api` 路由让特权方法集(`host.pickDirectory``host.openPath`,以及整个配置面——`settings.describe`/`openDocument`/`update`/`replace`/`mutate``credentials.describe`/`set`/`unset`;读取与原生操作也在内,因为 describe 会返回已暴露的配置、打开操作会作用于 Host 桌面,而探测任意引用会报出某条凭据来自何处)以空信任表过信任 fence从而钉在回环——已声明的 `trustedHosts` 授权可达其余全部方法,而这些方法在真正的认证层出现之前仍只限回环本机。平台载体与 ConnectionController 循环属于包内部apply 负责选择并驱动它们。下行边界见 [WebSocket 下行载体 Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md);协议契约见 api-contracts v3 §3。
协议消费层:客户端插件的 apply 会挂载 `ctx.connection`(共享 API 客户端 + 当前页面的 loopback 状态 + 单消费方流循环启动器);导出表层携带协议契约类型、`AbstractApiClient` seam以及循环的 sink配置类型。浏览器载体以 HTTP POST 发送 unaryrespond并为 `events.mux``events.host` 各开一条只下行的 WebSocket进程内载体满足同一双流抽象。Host half 持有唯一 `/api` route 及其 Fetch bridge已注册的 TypeRT interceptor 会先认领自己的 Remote endpoint未认领请求再回退 API Proxy。Loopback hostname 判定逻辑留在包内部:`/api` Host fence 与 WebSocket upgrade 会直接使用它,其他客户端插件则消费派生的 `ctx.connection.isLoopback` 状态。node 半侧的 `/api` 路由让特权方法集(`host.pickDirectory``host.openPath`,以及整个配置面——`settings.describe`/`openDocument`/`update`/`replace`/`mutate``credentials.describe`/`set`/`unset`;读取与原生操作也在内,因为 describe 会返回已暴露的配置、打开操作会作用于 Host 桌面,而探测任意引用会报出某条凭据来自何处)以空信任表过信任 fence从而钉在回环——已声明的 `trustedHosts` 授权可达其余全部方法,而这些方法在真正的认证层出现之前仍只限回环本机。平台载体与 ConnectionController 循环属于包内部apply 负责选择并驱动它们。下行边界见 [WebSocket 下行载体 Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md);协议契约见 api-contracts v3 §3。
## /api 浏览器信任栅栏

View File

@@ -16,12 +16,13 @@
import type { IncomingHttpHeaders } from 'node:http'
import { isLoopbackHostname } from './loopback-hostname.ts'
/** The request facts the fence reads (structural subset of IncomingMessage). */
/** The request facts the fence reads from either HTTP representation. */
interface ApiTrustRequest {
headers: IncomingHttpHeaders
headers: IncomingHttpHeaders | Headers
}
function header(headers: IncomingHttpHeaders, name: string): string | undefined {
function header(headers: IncomingHttpHeaders | Headers, name: string): string | undefined {
if (headers instanceof Headers) return headers.get(name) ?? undefined
const value = headers[name]
return typeof value === 'string' ? value : undefined
}
@@ -88,7 +89,7 @@ function isTrustedAuthority(hostUrl: URL, trustedHosts: readonly string[]): bool
/**
* Decide whether one /api request may reach the RPC bridge.
* @param request - node HTTP request facts (headers).
* @param request - Node HTTP or Fetch request facts (headers).
* @param trustedHosts - non-loopback authorities this deployment serves: exact `host:port`, or port-less `host` matching any port.
* @returns true when the Host is ours (loopback or trusted) and any attached browser markers are same-origin.
*/

View File

@@ -35,10 +35,12 @@ import type {
} from './api.ts'
import type { RequestPayload, ResponseValue, RpcMethodMap } from '@deepseek-ai/dsh-host-apiproxy/api'
import { AbstractApiClient, RpcId, SESSION_SEARCH_RESULT_LIMIT } from './api.ts'
import { randomUuid } from './random-uuid.ts'
import type { ClientConnectionRpc } from '../rpc.ts'
/** The fake carrier mints like a real one (business code never mints). */
function rpcRequest<P>(payload: P): RpcRequest<P> {
return { rpcId: RpcId(crypto.randomUUID()), payload }
return { rpcId: RpcId(randomUuid()), payload }
}
function text(t: string): ContentBlock[] {
@@ -1328,6 +1330,16 @@ class FxInbox<F> implements StreamConn<F> {
* @returns an ApiProxy backed entirely by in-memory state — no host process, no network.
*/
export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
return createFixtureWorld(options).api
}
interface FixtureWorld {
readonly api: ApiProxy
readonly rpc: ClientConnectionRpc
}
/** Build the fixture's legacy API and Remote RPC faces over one state graph. */
function createFixtureWorld(options: FixtureOptions): FixtureWorld {
// The resident fixture sessions all carry history, so none of them is blank.
const sessions: SessionSummary[] = options.empty ? [] : [
{ sessionId: sid('fx-alpha'), updatedAt: Date.now(), running: true, blank: false, cwd: '/tmp/fixture' },
@@ -1506,31 +1518,141 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
return backscanGoal(log) as FxGoalProjection
}
/** Shared CAS mutation path of the goal verbs (undefined next = invalid transition). */
const fxMutateGoal = (
request: RpcRequest<{ sessionId: SessionId; ref: { id: string; revision: number } }>,
ref: { id: string; revision: number },
next: (current: FxGoalProjection) => FxGoalProjection['goal'] | undefined,
): Promise<RpcResponse<{ ref: { id: never; revision: number } }>> => {
const missing = requireSession(request)
type FxGoalRef = { id: string; revision: number }
type FxGoalView = FxGoalProjection['goal'] & {
roundsStarted: number
createdAt: number
updatedAt: number
activation: 'armed' | 'disarmed'
}
const goalFailure = <T>(message: string): RpcResult<T> => ({
ok: false,
error: { code: 'internal', message, details: {} },
})
const requireGoalSession = (id: SessionId): RpcResult<never> | undefined => (
summaryOf(id) === undefined
? { ok: false, error: { code: 'session-not-found', message: `no session ${id}`, details: { sessionId: id } } }
: undefined
)
const goalView = (projection: FxGoalProjection): FxGoalView => ({
...projection.goal,
roundsStarted: projection.roundsStarted,
createdAt: projection.createdAt,
updatedAt: projection.updatedAt,
activation: projection.goal.phase === 'active' ? 'armed' : 'disarmed',
})
/** Canonical fixture implementation of the generated Goal Remote contract. */
const goalRemotes = {
create(id: SessionId, request: { objective: string; maxGoalRounds?: number }): RpcResult<{ ref: FxGoalRef }> {
const missing = requireGoalSession(id)
if (missing !== undefined) return missing
const current = backscanGoal(logOf(id))
if (current !== null && current.goal.phase !== 'complete') {
return goalFailure(`goal "${current.goal.id}" already exists`)
}
const now = Date.now()
const projection = appendGoalChange(id, {
kind: 'goal/change', version: 1, operation: 'create',
goal: {
id: `fx-goal-${logOf(id).length}`,
revision: 1,
objective: request.objective,
phase: 'active',
maxGoalRounds: request.maxGoalRounds ?? 256,
},
roundsStarted: 0, createdAt: now, updatedAt: now,
})
return { ok: true, value: { ref: { id: projection.goal.id, revision: projection.goal.revision } } }
},
edit(id: SessionId, ref: FxGoalRef, request: { objective?: string; maxGoalRounds?: number }): RpcResult<FxGoalView> {
return mutateGoal(id, ref, current => ({
...current.goal,
revision: current.goal.revision + 1,
...request.objective === undefined ? {} : { objective: request.objective },
...request.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.maxGoalRounds },
}))
},
pause(id: SessionId, ref: FxGoalRef): RpcResult<FxGoalView> {
return mutateGoal(id, ref, current => (
current.goal.phase === 'active'
? { ...current.goal, revision: current.goal.revision + 1, phase: 'paused' }
: undefined
))
},
resume(id: SessionId, ref: FxGoalRef): RpcResult<FxGoalView> {
return mutateGoal(id, ref, current => (
current.goal.phase === 'paused' || current.goal.phase === 'blocked' || current.goal.phase === 'active'
? { ...current.goal, revision: current.goal.revision + 1, phase: 'active' }
: undefined
))
},
complete(id: SessionId, ref: FxGoalRef): RpcResult<FxGoalView> {
return mutateGoal(id, ref, current => (
current.goal.phase === 'complete'
? undefined
: { ...current.goal, revision: current.goal.revision + 1, phase: 'complete' }
))
},
clear(id: SessionId, ref: FxGoalRef): RpcResult<FxGoalRef> {
const resolved = resolveGoal(id, ref)
if (!resolved.ok) return resolved
const current = resolved.value
const tombstone = { id: current.goal.id, revision: current.goal.revision + 1 }
appendGoalChange(id, {
kind: 'goal/change', version: 1, operation: 'clear', cleared: tombstone, clearedAt: Date.now(),
})
return { ok: true, value: tombstone }
},
}
/** Resolve one current goal revision for a canonical Remote mutation. */
function resolveGoal(id: SessionId, ref: FxGoalRef): RpcResult<FxGoalProjection> {
const missing = requireGoalSession(id)
if (missing !== undefined) return missing
const id = request.payload.sessionId
const current = backscanGoal(logOf(id))
if (current === null || current.goal.id !== ref.id || current.goal.revision !== ref.revision) {
return err(request, { code: 'internal', message: 'stale or missing goal revision', details: { goalCode: 'GOAL_STALE_REVISION' } })
return goalFailure('stale or missing goal revision')
}
return { ok: true, value: current }
}
/** Shared CAS mutation path behind the canonical Remote verbs. */
function mutateGoal(
id: SessionId,
ref: FxGoalRef,
next: (current: FxGoalProjection) => FxGoalProjection['goal'] | undefined,
): RpcResult<FxGoalView> {
const resolved = resolveGoal(id, ref)
if (!resolved.ok) return resolved
const current = resolved.value
const goal = next(current)
if (goal === undefined) {
return err(request, { code: 'internal', message: `invalid goal transition from "${current.goal.phase}"`, details: { goalCode: 'GOAL_INVALID_TRANSITION' } })
return goalFailure(`invalid goal transition from "${current.goal.phase}"`)
}
const projection = appendGoalChange(id, {
kind: 'goal/change', version: 1,
operation: goal.phase === current.goal.phase ? 'edit' : goal.phase === 'paused' ? 'pause' : goal.phase === 'active' ? 'resume' : 'complete',
goal, roundsStarted: current.roundsStarted, createdAt: current.createdAt, updatedAt: Date.now(),
})
return ok(request, { ref: { id: projection.goal.id as never, revision: projection.goal.revision } })
return { ok: true, value: goalView(projection) }
}
const mapGoalResult = <T, U>(result: RpcResult<T>, map: (value: T) => U): RpcResult<U> => (
result.ok ? { ok: true, value: map(result.value) } : result
)
const goalRefResult = (result: RpcResult<FxGoalView>): RpcResult<{ ref: { id: never; revision: number } }> => (
mapGoalResult(result, view => ({ ref: { id: view.id as never, revision: view.revision } }))
)
const legacyGoalResponse = <P, T>(request: RpcRequest<P>, result: RpcResult<T>): Promise<RpcResponse<T>> => (
Promise.resolve({ rpcId: request.rpcId, result })
)
/** At most one in-flight replay per session; cancel clears it. */
const replays = new Map<SessionId, { timer: ReturnType<typeof setTimeout>; finish(aborted: boolean): void }>()
@@ -1776,7 +1898,7 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
replays.set(id, { timer: setTimeout(tick, 80), finish })
}
return {
const api: ApiProxy = {
sessions: {
list: request => ok(request, { items: [...sessions].sort((a, b) => b.updatedAt - a.updatedAt) }),
search: (request, signal) => {
@@ -1978,6 +2100,9 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
models: request => ok(request, {
current: modelTargets.get(request.payload.sessionId)
?? { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
// The fixture's routes all serve; a surface exercising the blocked
// posture drives it through its own stub.
routable: true,
groups: fixtureModelGroups(),
failures: [],
}),
@@ -2342,60 +2467,44 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
},
},
goals: {
// Mutation-only mirror of the host handlers: each verb CAS-checks the
// projected current goal, appends the whole-value change (the mux
// stream and projection frame ride the shared append path), and
// acknowledges with the new ref only.
create: (request) => {
const missing = requireSession(request)
if (missing !== undefined) return missing
const id = request.payload.sessionId
const current = backscanGoal(logOf(id))
if (current !== null && current.goal.phase !== 'complete') {
return err(request, { code: 'internal', message: `goal "${current.goal.id}" already exists`, details: { goalCode: 'GOAL_ALREADY_EXISTS' } })
}
const projection = appendGoalChange(id, {
kind: 'goal/change', version: 1, operation: 'create',
goal: { id: `fx-goal-${logOf(id).length}`, revision: 1, objective: request.payload.objective, phase: 'active', maxGoalRounds: request.payload.maxGoalRounds ?? 256 },
roundsStarted: 0, createdAt: Date.now(), updatedAt: Date.now(),
})
return ok(request, { ref: { id: projection.goal.id as never, revision: projection.goal.revision } })
},
edit: request => fxMutateGoal(request, request.payload.ref, current => ({
...current.goal,
revision: current.goal.revision + 1,
...request.payload.objective === undefined ? {} : { objective: request.payload.objective },
...request.payload.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.payload.maxGoalRounds },
})),
pause: request => fxMutateGoal(request, request.payload.ref, current => (
current.goal.phase === 'active'
? { ...current.goal, revision: current.goal.revision + 1, phase: 'paused' }
: undefined
)),
resume: request => fxMutateGoal(request, request.payload.ref, current => (
current.goal.phase === 'paused' || current.goal.phase === 'blocked' || current.goal.phase === 'active'
? { ...current.goal, revision: current.goal.revision + 1, phase: 'active' }
: undefined
)),
complete: request => fxMutateGoal(request, request.payload.ref, current => (
current.goal.phase === 'complete'
? undefined
: { ...current.goal, revision: current.goal.revision + 1, phase: 'complete' }
)),
clear: (request) => {
const missing = requireSession(request)
if (missing !== undefined) return missing
const id = request.payload.sessionId
const current = backscanGoal(logOf(id))
if (current === null || current.goal.id !== request.payload.ref.id || current.goal.revision !== request.payload.ref.revision) {
return err(request, { code: 'internal', message: 'stale or missing goal revision', details: { goalCode: 'GOAL_STALE_REVISION' } })
}
appendGoalChange(id, {
kind: 'goal/change', version: 1, operation: 'clear',
cleared: { id: current.goal.id, revision: current.goal.revision + 1 }, clearedAt: Date.now(),
})
return ok(request, { cleared: true as const })
},
// Compatibility face only: old API Proxy payloads and acknowledgements
// adapt to the canonical fixture Remote implementation above.
create: request => legacyGoalResponse(
request,
mapGoalResult(
goalRemotes.create(request.payload.sessionId, {
objective: request.payload.objective,
...request.payload.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.payload.maxGoalRounds },
}),
value => ({ ref: { id: value.ref.id as never, revision: value.ref.revision } }),
),
),
edit: request => legacyGoalResponse(
request,
goalRefResult(goalRemotes.edit(request.payload.sessionId, request.payload.ref, {
...request.payload.objective === undefined ? {} : { objective: request.payload.objective },
...request.payload.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.payload.maxGoalRounds },
})),
),
pause: request => legacyGoalResponse(
request,
goalRefResult(goalRemotes.pause(request.payload.sessionId, request.payload.ref)),
),
resume: request => legacyGoalResponse(
request,
goalRefResult(goalRemotes.resume(request.payload.sessionId, request.payload.ref)),
),
complete: request => legacyGoalResponse(
request,
goalRefResult(goalRemotes.complete(request.payload.sessionId, request.payload.ref)),
),
clear: request => legacyGoalResponse(
request,
mapGoalResult(
goalRemotes.clear(request.payload.sessionId, request.payload.ref),
() => ({ cleared: true as const }),
),
),
},
events: {
async *mux(_request, signal) {
@@ -2515,8 +2624,11 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
providers: request => ok(request, {
providers: [
{ provider: 'deepseek-official', displayName: 'DeepSeek', settingsNs: 'llm-deepseek', settingsPath: [], active: true },
{ provider: 'openai', displayName: 'openai', settingsNs: 'llm-pi-ai', settingsPath: ['providers', 'openai'], active: true },
{ provider: 'anthropic', displayName: 'anthropic', settingsNs: 'llm-pi-ai', settingsPath: ['providers', 'anthropic'], active: false },
{ provider: 'openai', displayName: 'openai', settingsNs: 'llm-pi-ai', settingsPath: ['providers', 'openai'], active: true, declared: false },
{ provider: 'anthropic', displayName: 'anthropic', settingsNs: 'llm-pi-ai', settingsPath: ['providers', 'anthropic'], active: false, declared: false },
// One hand-declared route, so a surface reading this fixture meets
// the tagged shape rather than only the shipped one.
{ provider: 'acme-gateway', displayName: 'Acme Gateway', settingsNs: 'llm-pi-ai', settingsPath: ['providers', 'acme-gateway'], active: true, declared: true },
],
}),
models: request => ok(request, { groups: fixtureModelGroups(), failures: [] }),
@@ -2553,6 +2665,36 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
return Promise.resolve({ accepted: true })
},
}
const rpc: ClientConnectionRpc = {
call(channel, endpoint, payload) {
if (channel !== '/api') {
return Promise.reject(new Error(`fixture connection RPC channel ${JSON.stringify(channel)} is unavailable`))
}
const args = (payload as {
args: {
agentId: SessionId
ref?: { id: string; revision: number }
request?: { objective?: string; maxGoalRounds?: number }
}
}).args
const sessionId = args.agentId
switch (endpoint) {
case 'goals/create': return Promise.resolve(goalRemotes.create(sessionId, {
objective: args.request?.objective as string,
...args.request?.maxGoalRounds === undefined ? {} : { maxGoalRounds: args.request.maxGoalRounds },
}))
case 'goals/edit': return Promise.resolve(goalRemotes.edit(sessionId, args.ref as FxGoalRef, args.request ?? {}))
case 'goals/pause': return Promise.resolve(goalRemotes.pause(sessionId, args.ref as FxGoalRef))
case 'goals/resume': return Promise.resolve(goalRemotes.resume(sessionId, args.ref as FxGoalRef))
case 'goals/complete': return Promise.resolve(goalRemotes.complete(sessionId, args.ref as FxGoalRef))
case 'goals/clear': return Promise.resolve(goalRemotes.clear(sessionId, args.ref as FxGoalRef))
default:
return Promise.reject(new Error(`fixture connection RPC endpoint ${JSON.stringify(endpoint)} is unavailable`))
}
},
}
return { api, rpc }
}
/**
@@ -2564,10 +2706,14 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
*/
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()
this.api = createFixtureApi(fixtureOptionsFromLocation())
const world = createFixtureWorld(fixtureOptionsFromLocation())
this.api = world.api
this.rpc = world.rpc
}
protected doFetch(): Promise<Response> {

View File

@@ -8,7 +8,9 @@ import type { IApiClient } from './api.ts'
import { ConnectionController, type ConnectionConfig, type ConnectionSinks, type ConnectionState } from './connection.ts'
import { FixtureApiClient } from './fixture.ts'
import { WebApiClient } from './web-api-client.ts'
import { createWebConnectionRpc } from './rpc.ts'
import { isLoopbackHostname } from '../loopback-hostname.ts'
import type { ClientConnectionRpc } from '../rpc.ts'
// ---- Contract re-exports (browser-safe apiproxy channels + core types) ----
export type {
@@ -36,6 +38,7 @@ export {
// Connection loop types are public through ConnectionHandle.start; the
// controller remains package-internal.
export type { ConnectionConfig, ConnectionSinks, ConnectionState }
export type { ClientConnectionRpc } from '../rpc.ts'
/** Required services (none — this is the wire root). */
@@ -51,6 +54,8 @@ export interface ConnectionHandle {
readonly api: IApiClient
/** Whether the current page authority is loopback; non-browser contexts default to true. */
readonly isLoopback: boolean
/** Generic logical RPC channels over the same Connection transport. */
readonly rpc: ClientConnectionRpc
/**
* Start the connect/pump/reconnect loop with the consumer's frame sinks.
* One consumer owns the streams (the runtime object layer); a second call
@@ -69,11 +74,14 @@ export interface ConnectionHandle {
export function apply(ctx: Context): void {
const pageLocation = typeof location === 'undefined' ? undefined : location
const fixture = pageLocation !== undefined && new URLSearchParams(pageLocation.search).has('fixture')
const api: IApiClient = fixture ? new FixtureApiClient() : new WebApiClient()
const fixtureClient = fixture ? new FixtureApiClient() : undefined
const api: IApiClient = fixtureClient ?? new WebApiClient()
const rpc = fixtureClient?.rpc ?? createWebConnectionRpc()
let started = false
const handle: ConnectionHandle = {
api,
isLoopback: pageLocation === undefined || isLoopbackHostname(pageLocation.hostname),
rpc,
start(sinks, config) {
if (started) throw new Error('connection: the stream loop is already owned by another consumer')
started = true

View File

@@ -0,0 +1,14 @@
/** Browser-safe UUID generation for client-side wire correlation. */
/**
* Generate an RFC 4122 version 4 UUID without requiring a secure context.
* @returns a UUID backed by `crypto.getRandomValues()`, which browsers expose on insecure origins.
*/
export function randomUuid(): string {
const bytes = globalThis.crypto.getRandomValues(new Uint8Array(16))
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
view.setUint8(6, (view.getUint8(6) & 0x0f) | 0x40)
view.setUint8(8, (view.getUint8(8) & 0x3f) | 0x80)
const hex = Array.from(bytes, byte => byte.toString(16).padStart(2, '0')).join('')
return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`
}

View File

@@ -0,0 +1,63 @@
/** Browser caller for generic Connection unary RPC channels. */
import {
RpcId,
serverResponseSchema,
type ClientRequest,
} from '@deepseek-ai/dsh-host-apiproxy/api'
import type { ClientConnectionRpc } from '../rpc.ts'
import { randomUuid } from './random-uuid.ts'
const INTERNAL_BASE = 'http://dsh.internal'
const CHANNEL_PATTERN = /^\/[A-Za-z0-9._~-]+$/
const ENDPOINT_SEGMENT_PATTERN = /^[A-Za-z0-9_$.-]+$/
/**
* Create the browser-backed generic RPC caller.
* @returns caller that owns request correlation and response-envelope validation.
*/
export function createWebConnectionRpc(): ClientConnectionRpc {
return {
async call(channel, endpoint, payload, signal) {
assertTarget(channel, endpoint)
const rpcId = RpcId(randomUuid())
const message: ClientRequest = {
type: 'client-request',
rpcId,
method: endpoint,
payload,
}
const response = await globalThis.fetch(
new URL(`${channel}/${endpoint}`, resolveBase()),
{
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify(message),
...signal === undefined ? {} : { signal },
},
)
if (!response.ok) {
throw new Error(`transport failure for ${channel}/${endpoint}: HTTP ${response.status}`)
}
const full = serverResponseSchema.parse(await response.json())
if (full.rpcId !== rpcId) {
throw new Error(`rpcId mismatch for ${endpoint}: sent ${rpcId}, got ${full.rpcId}`)
}
return full.result
},
}
}
function resolveBase(): string {
const location = (globalThis as { location?: { origin?: string } }).location
return location?.origin !== undefined && location.origin !== 'null' ? location.origin : INTERNAL_BASE
}
function assertTarget(channel: string, endpoint: string): void {
const segments = endpoint.split('/')
if (!CHANNEL_PATTERN.test(channel)
|| segments.some(segment =>
segment === '' || segment === '.' || segment === '..' || !ENDPOINT_SEGMENT_PATTERN.test(segment))) {
throw new Error(`connection: invalid RPC target ${JSON.stringify(`${channel}/${endpoint}`)}`)
}
}

View File

@@ -5,6 +5,16 @@
import type { IncomingMessage, ServerResponse } from 'node:http'
/** Transport-independent request handler consumed by the Host HTTP bridge. */
export interface FetchHandler {
/**
* Handle one standard Fetch request.
* @param request - request produced by the active transport bridge.
* @returns complete or streaming Fetch response.
*/
fetch(request: Request): Promise<Response>
}
/**
* Bridge one node:http request to the fetch-shaped handler (client close
* aborts; SSE bodies stream out chunk by chunk).
@@ -12,7 +22,7 @@ import type { IncomingMessage, ServerResponse } from 'node:http'
* @param res - node:http response the bridge writes and owns to completion.
* @param apiHandler - fetch-shaped API carrier the request is dispatched to.
*/
export async function bridge(req: IncomingMessage, res: ServerResponse, apiHandler: { fetch: typeof fetch }): Promise<void> {
export async function bridge(req: IncomingMessage, res: ServerResponse, apiHandler: FetchHandler): Promise<void> {
const abort = new AbortController()
// Client-disconnect detection MUST hang off the response, not the request:
// since Node 16, IncomingMessage 'close' fires as soon as the request body is

View File

@@ -7,15 +7,26 @@ import { toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
import { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'
import { bridge } from './http-bridge.ts'
import { assertTrustedAuthority, isTrustedApiRequest } from './api-request-trust.ts'
import { HostConnectionService } from './rpc-host.ts'
import { rejectWebSocketUpgrade, WebSocketDownlinks } from './websocket-downlink.ts'
export type {
ConnectionRpcAuthority,
ConnectionRpcEndpointMatcher,
ConnectionRpcHandler,
ConnectionRpcHandlerOptions,
HostConnectionHandle,
HostConnectionRpc,
} from './rpc.ts'
export { HostConnectionService } from './rpc-host.ts'
export { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'
/** Stable Cordis plugin name. */
export const name = 'client-connection'
/** Services required before mounting the route. */
export const inject = ['httpServer', 'apiProxy']
/** Services required before providing Connection; API Proxy is an optional `/api` fallback. */
export const inject = ['httpServer']
/** Plugin config: the deployment's non-loopback serving authorities. */
export interface ConnectionConfig {
@@ -83,49 +94,61 @@ export function apply(ctx: Context, config?: ConnectionConfig): void {
// 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)
const apiHandler = toFetchHandler(ctx.apiProxy)
const downlinks = new WebSocketDownlinks(ctx.apiProxy)
const connection = new HostConnectionService(ctx, trustedHosts)
const fetchHandler = connection.createSharedFetchHandler(API_PATH, {
async fetch(request) {
const pathname = new URL(request.url).pathname
const method = pathname.startsWith(`${API_PATH}/`)
? pathname.slice(API_PATH.length + 1)
: undefined
if (method !== undefined
&& PRIVILEGED_METHODS.has(method)
&& !isTrustedApiRequest(request, [])) {
return new Response('forbidden', { status: 403 })
}
if (request.method === 'GET' && (pathname === MUX_EVENTS_PATH || pathname === HOST_EVENTS_PATH)) {
return new Response('upgrade required', {
status: 426,
headers: { connection: 'Upgrade', upgrade: 'websocket' },
})
}
const apiProxy = ctx.get('apiProxy')
if (apiProxy === undefined) return new Response('not found', { status: 404 })
return toFetchHandler(apiProxy).fetch(request)
},
})
const route: WebRoute = {
kind: 'prefix',
path: API_PATH,
handler: async (req, res) => {
const pathname = new URL(req.url ?? '/', 'http://dsh.internal').pathname
const method = pathname.startsWith(`${API_PATH}/`)
? pathname.slice(API_PATH.length + 1)
: undefined
const allowed = method !== undefined && PRIVILEGED_METHODS.has(method)
? isTrustedApiRequest(req, [])
: isTrustedApiRequest(req, trustedHosts)
if (!allowed) {
if (!isTrustedApiRequest(req, trustedHosts)) {
res.writeHead(403)
res.end('forbidden')
return
}
if (req.method === 'GET' && (pathname === MUX_EVENTS_PATH || pathname === HOST_EVENTS_PATH)) {
res.writeHead(426, { connection: 'Upgrade', upgrade: 'websocket' })
res.end('upgrade required')
return
}
await bridge(req, res, apiHandler)
await bridge(req, res, fetchHandler)
},
}
ctx.effect(() => ctx.httpServer.register(route), 'client-connection: /api route')
const registerDownlink = (
path: string,
handle: WebUpgradeRoute['handler'],
): void => {
ctx.effect(() => ctx.httpServer.registerUpgrade({
path,
handler: (req, socket, head) => {
if (!isTrustedApiRequest(req, trustedHosts)) {
rejectWebSocketUpgrade(socket)
return
}
return handle(req, socket, head)
},
}), `client-connection: ${path} WebSocket`)
}
ctx.effect(() => () => downlinks.close(), 'client-connection: WebSocket downlinks')
registerDownlink(MUX_EVENTS_PATH, (req, socket, head) => { downlinks.handleMux(req, socket, head) })
registerDownlink(HOST_EVENTS_PATH, (req, socket, head) => { downlinks.handleHost(req, socket, head) })
ctx.inject(['apiProxy'], (apiCtx) => {
const downlinks = new WebSocketDownlinks(apiCtx.apiProxy)
const registerDownlink = (
path: string,
handle: WebUpgradeRoute['handler'],
): void => {
apiCtx.effect(() => apiCtx.httpServer.registerUpgrade({
path,
handler: (req, socket, head) => {
if (!isTrustedApiRequest(req, trustedHosts)) {
rejectWebSocketUpgrade(socket)
return
}
return handle(req, socket, head)
},
}), `client-connection: ${path} WebSocket`)
}
apiCtx.effect(() => () => downlinks.close(), 'client-connection: WebSocket downlinks')
registerDownlink(MUX_EVENTS_PATH, (req, socket, head) => { downlinks.handleMux(req, socket, head) })
registerDownlink(HOST_EVENTS_PATH, (req, socket, head) => { downlinks.handleHost(req, socket, head) })
})
}

View File

@@ -0,0 +1,224 @@
/** Host registry and HTTP adapter for generic Connection RPC channels. */
import { Context, Service } from 'cordis'
import type { WebRoute } from '@deepseek-ai/dsh-host-webserver'
import {
clientRequestSchema,
RpcId,
type ClientRequest,
type RpcError,
type RpcErrorDetailsMap,
type RpcId as RpcIdType,
type ServerResponse as RpcServerResponse,
} from '@deepseek-ai/dsh-host-apiproxy/api'
import { bridge, type FetchHandler } from './http-bridge.ts'
import { isTrustedApiRequest } from './api-request-trust.ts'
import { API_PATH } from './api-path.ts'
import type {
ConnectionRpcEndpointMatcher,
ConnectionRpcHandler,
ConnectionRpcHandlerOptions,
HostConnectionHandle,
HostConnectionRpc,
} from './rpc.ts'
const INVALID_REQUEST_RPC_ID = RpcId('invalid-request')
const CHANNEL_PATTERN = /^\/[A-Za-z0-9._~-]+$/
const ENDPOINT_SEGMENT_PATTERN = /^[A-Za-z0-9_$.-]+$/
interface ConnectionRpcInterceptor {
readonly matches: ConnectionRpcEndpointMatcher
readonly fetchHandler: FetchHandler
readonly options: ConnectionRpcHandlerOptions
}
declare module 'cordis' {
interface Context {
/** Host Connection transport and RPC registrations. */
connection: HostConnectionHandle
}
}
/** Host Connection service whose channel registrations belong to the caller fiber. */
export class HostConnectionService extends Service implements HostConnectionHandle {
private readonly interceptors = new Map<string, ConnectionRpcInterceptor>()
/**
* Provide the Host half over the active HTTP server.
* @param ctx - owning Connection plugin context.
* @param trustedHosts - deployment authorities accepted by trusted-host channels.
*/
constructor(ctx: Context, private readonly trustedHosts: readonly string[]) {
super(ctx, 'connection')
}
/** Generic channel registry scoped to the Context reading this service. */
get rpc(): HostConnectionRpc {
const owner = this.ctx
return {
handle: (channel, handler, options) => this.register(owner, channel, handler, options),
intercept: (channel, matches, handler, options) =>
this.registerInterceptor(owner, channel, matches, handler, options),
}
}
/**
* Compose one shared-channel Fetch handler from its interceptor and fallback.
* @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.
*/
createSharedFetchHandler(
channel: '/api',
fallback: FetchHandler,
): FetchHandler {
return {
fetch: (request) => {
const endpoint = endpointFromPath(channel, new URL(request.url).pathname)
const interceptor = this.interceptors.get(channel)
if (endpoint === undefined || interceptor === undefined || !interceptor.matches(endpoint)) {
return fallback.fetch(request)
}
if (interceptor.options.authority === 'loopback' && !isTrustedApiRequest(request, [])) {
return Promise.resolve(new Response('forbidden', { status: 403 }))
}
return interceptor.fetchHandler.fetch(request)
},
}
}
private register(
owner: Context,
channel: string,
handler: ConnectionRpcHandler,
options: ConnectionRpcHandlerOptions,
): () => Promise<void> {
assertChannel(channel)
const trustedHosts = options.authority === 'loopback' ? [] : this.trustedHosts
const fetchHandler = rpcFetchHandler(channel, handler)
const route: WebRoute = {
kind: 'prefix',
path: channel,
handler: async (req, res) => {
if (!isTrustedApiRequest(req, trustedHosts)) {
res.writeHead(403)
res.end('forbidden')
return
}
await bridge(req, res, fetchHandler)
},
}
return owner.effect(
() => owner.httpServer.register(route),
`client-connection: ${channel} rpc channel`,
)
}
private registerInterceptor(
owner: Context,
channel: string,
matches: ConnectionRpcEndpointMatcher,
handler: ConnectionRpcHandler,
options: ConnectionRpcHandlerOptions,
): () => Promise<void> {
if (channel !== API_PATH) {
throw new Error(`connection: invalid shared RPC channel ${JSON.stringify(channel)}`)
}
const interceptor: ConnectionRpcInterceptor = {
matches,
fetchHandler: rpcFetchHandler(channel, handler),
options,
}
return owner.effect(() => {
if (this.interceptors.has(channel)) {
throw new Error(`connection: shared RPC channel ${JSON.stringify(channel)} already has an interceptor`)
}
this.interceptors.set(channel, interceptor)
return () => {
this.interceptors.delete(channel)
}
}, `client-connection: ${channel} rpc interceptor`)
}
}
function rpcFetchHandler(
channel: string,
handler: ConnectionRpcHandler,
): FetchHandler {
return {
async fetch(request: Request): Promise<Response> {
const endpoint = endpointFromPath(channel, new URL(request.url).pathname)
if (request.method !== 'POST' || endpoint === undefined) {
return new Response('not found', { status: 404 })
}
const mediaType = request.headers.get('content-type')?.split(';', 1)[0]?.trim().toLowerCase()
if (mediaType !== 'application/json') {
return new Response('content type must be application/json', { status: 415 })
}
let body: unknown
try {
body = await request.json()
} catch {
return new Response('body is not JSON', { status: 400 })
}
const envelope = clientRequestSchema.safeParse(body)
if (!envelope.success) {
return invalidEnvelopeResponse(body, envelope.error.issues)
}
const message: ClientRequest = envelope.data
if (message.method !== endpoint) {
return errorResponse(message.rpcId, {
code: 'bad-request',
message: `method ${JSON.stringify(message.method)} does not match endpoint ${JSON.stringify(endpoint)}`,
details: { issues: [] },
})
}
try {
const result = await handler(endpoint, message.payload, request.signal)
return fullResponse(message.rpcId, result)
} catch (error) {
return new Response(`handler failure: ${String(error)}`, { status: 500 })
}
},
}
}
function invalidEnvelopeResponse(body: unknown, issues: RpcErrorDetailsMap['bad-request']['issues']): Response {
const rawId = (body as { rpcId?: unknown } | null)?.rpcId
const rpcId = typeof rawId === 'string' ? RpcId(rawId) : INVALID_REQUEST_RPC_ID
return errorResponse(rpcId, {
code: 'bad-request',
message: 'invalid client-request message',
details: { issues },
})
}
function endpointFromPath(channel: string, pathname: string): string | undefined {
if (!pathname.startsWith(`${channel}/`)) return undefined
const endpoint = pathname.slice(channel.length + 1)
const segments = endpoint.split('/')
if (segments.some(segment =>
segment === '' || segment === '.' || segment === '..' || !ENDPOINT_SEGMENT_PATTERN.test(segment))) {
return undefined
}
return endpoint
}
function errorResponse(rpcId: RpcIdType, error: RpcError): Response {
return fullResponse(rpcId, { ok: false, error })
}
function fullResponse(rpcId: RpcIdType, result: RpcServerResponse['result']): Response {
const body: RpcServerResponse = { type: 'server-response', rpcId, result }
return Response.json(body)
}
function assertChannel(channel: string): void {
if (!CHANNEL_PATTERN.test(channel) || channel === '/api') {
throw new Error(`connection: invalid or reserved RPC channel ${JSON.stringify(channel)}`)
}
}

View File

@@ -0,0 +1,77 @@
/** Generic unary RPC contracts shared by the Host and Client Connection halves. */
import type { RpcResult } from '@deepseek-ai/dsh-host-apiproxy/api'
/** Trust fence applied before a Host RPC channel reaches its handler. */
export type ConnectionRpcAuthority = 'trusted-host' | 'loopback'
/** Registration policy for one logical RPC channel. */
export interface ConnectionRpcHandlerOptions {
/** Browser authority accepted by every endpoint in this channel. */
readonly authority: ConnectionRpcAuthority
}
/** Handler invoked after Connection has decoded the transport envelope. */
export type ConnectionRpcHandler = (
endpoint: string,
payload: unknown,
signal: AbortSignal,
) => Promise<RpcResult<unknown>>
/** Synchronous ownership test for one endpoint on a shared RPC channel. */
export type ConnectionRpcEndpointMatcher = (endpoint: string) => boolean
/** Host registry for logical RPC channels carried by the current transport. */
export interface HostConnectionRpc {
/**
* Register one absolute channel prefix and its trust policy.
* @param channel - absolute logical channel such as `/rpc`.
* @param handler - decoded endpoint handler returning the existing RPC result shape.
* @param options - channel trust policy.
* @returns asynchronous disposer removing the channel and its physical route.
*/
handle(
channel: string,
handler: ConnectionRpcHandler,
options: ConnectionRpcHandlerOptions,
): () => Promise<void>
/**
* Intercept owned endpoints on the shared `/api` channel before its fallback.
* @param channel - reserved shared channel; currently `/api`.
* @param matches - synchronous endpoint ownership test.
* @param handler - decoded endpoint handler returning the existing RPC result shape.
* @param options - trust policy for every endpoint claimed by this interceptor.
* @returns asynchronous disposer removing the interceptor.
*/
intercept(
channel: '/api',
matches: ConnectionRpcEndpointMatcher,
handler: ConnectionRpcHandler,
options: ConnectionRpcHandlerOptions,
): () => Promise<void>
}
/** Host `ctx.connection` shape consumed by transport-independent adapters. */
export interface HostConnectionHandle {
/** Generic RPC channel registry. */
readonly rpc: HostConnectionRpc
}
/** Client caller for logical RPC channels carried by the current transport. */
export interface ClientConnectionRpc {
/**
* Call one endpoint through an already registered logical channel.
* @param channel - absolute logical channel such as `/api`.
* @param endpoint - channel-relative endpoint such as `goals/create`.
* @param payload - channel-owned request payload.
* @param signal - optional caller cancellation.
* @returns the existing RPC success/error result; correlation stays inside Connection.
*/
call(
channel: string,
endpoint: string,
payload: unknown,
signal?: AbortSignal,
): Promise<RpcResult<unknown>>
}

View File

@@ -203,4 +203,119 @@ describe('connection client apply', () => {
expect(sockets).toHaveLength(1)
expect(sockets[0]?.readyState).toBe(FakeWebSocket.CLOSED)
})
it('carries RPC calls without requiring secure-context randomUUID', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '' }
vi.stubGlobal('crypto', {
getRandomValues(bytes: Uint8Array) {
return bytes.fill(0)
},
})
const handle = await mount()
const original = globalThis.fetch
const seen: { url: string; body: unknown }[] = []
globalThis.fetch = async (input: URL | RequestInfo, init?: RequestInit) => {
const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url
if (typeof init?.body !== 'string') throw new TypeError('expected a JSON string request body')
const body = JSON.parse(init.body) as { rpcId: string }
seen.push({ url, body })
return Response.json({
type: 'server-response',
rpcId: body.rpcId,
result: { ok: true, value: { ref: 'goal-1' } },
})
}
try {
await expect(handle.rpc.call('/api', 'goals/create', { args: { agentId: 'agent-1' } }))
.resolves.toEqual({ ok: true, value: { ref: 'goal-1' } })
} finally {
globalThis.fetch = original
vi.unstubAllGlobals()
}
expect(seen).toHaveLength(1)
expect(seen[0]?.url).toBe('http://dsh.internal/api/goals/create')
expect(seen[0]?.body).toMatchObject({
type: 'client-request',
rpcId: '00000000-0000-4000-8000-000000000000',
method: 'goals/create',
payload: { args: { agentId: 'agent-1' } },
})
})
it('validates generic RPC transport failures, correlation, and targets', async () => {
;(globalThis as Win).location = {
hostname: 'harness.example', search: '', origin: 'https://harness.example',
}
const handle = await mount()
const original = globalThis.fetch
const abort = new AbortController()
globalThis.fetch = vi.fn().mockResolvedValue(new Response('unavailable', { status: 503 }))
try {
await expect(handle.rpc.call('/api', 'goals/create', {}, abort.signal))
.rejects.toThrow('HTTP 503')
expect(globalThis.fetch).toHaveBeenCalledWith(
new URL('https://harness.example/api/goals/create'),
expect.objectContaining({ signal: abort.signal }),
)
;(globalThis as Win).location = { hostname: 'localhost', search: '', origin: 'null' }
globalThis.fetch = vi.fn().mockResolvedValue(Response.json({
type: 'server-response',
rpcId: 'different-rpc',
result: { ok: true, value: null },
}))
await expect(handle.rpc.call('/api', 'goals/create', {})).rejects.toThrow('rpcId mismatch')
const fetch = vi.mocked(globalThis.fetch)
expect(fetch.mock.calls[0]?.[0]).toEqual(new URL('http://dsh.internal/api/goals/create'))
expect(fetch.mock.calls[0]?.[1]).not.toHaveProperty('signal')
} finally {
globalThis.fetch = original
}
for (const [channel, endpoint] of [
['api2', 'goals/create'],
['/api/path', 'goals/create'],
['/api', ''],
['/api', '.'],
['/api', '..'],
['/api', 'goals//create'],
['/api', 'goals/create?unsafe'],
] as const) {
await expect(handle.rpc.call(channel, endpoint, {})).rejects.toThrow('invalid RPC target')
}
})
it('carries Goal Remotes over the same state as the client-only fixture API', async () => {
;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
const handle = await mount()
const created = await handle.rpc.call('/api', 'goals/create', {
args: { agentId: 'fx-alpha', request: { objective: 'fixture remote' } },
})
expect(created).toMatchObject({ ok: true, value: { ref: { revision: 1 } } })
if (!created.ok) throw new Error('fixture Goal create failed')
const ref = (created.value as { ref: { id: string; revision: number } }).ref
const edited = await handle.rpc.call('/api', 'goals/edit', {
args: { agentId: 'fx-alpha', ref, request: { objective: 'edited fixture remote' } },
})
expect(edited).toMatchObject({ ok: true, value: { objective: 'edited fixture remote', revision: 2 } })
const editedRef = { id: ref.id, revision: 2 }
const paused = await handle.rpc.call('/api', 'goals/pause', {
args: { agentId: 'fx-alpha', ref: editedRef },
})
expect(paused).toMatchObject({ ok: true, value: { phase: 'paused', activation: 'disarmed', revision: 3 } })
const resumed = await handle.rpc.call('/api', 'goals/resume', {
args: { agentId: 'fx-alpha', ref: { id: ref.id, revision: 3 } },
})
expect(resumed).toMatchObject({ ok: true, value: { phase: 'active', activation: 'armed', revision: 4 } })
const completed = await handle.rpc.call('/api', 'goals/complete', {
args: { agentId: 'fx-alpha', ref: { id: ref.id, revision: 4 } },
})
expect(completed).toMatchObject({ ok: true, value: { phase: 'complete', activation: 'disarmed', revision: 5 } })
await expect(handle.rpc.call('/api', 'goals/clear', {
args: { agentId: 'fx-alpha', ref: { id: ref.id, revision: 5 } },
})).resolves.toEqual({ ok: true, value: { id: ref.id, revision: 6 } })
await expect(handle.rpc.call('/other', 'goals/create', {})).rejects.toThrow(/channel.*unavailable/)
await expect(handle.rpc.call('/api', 'unknown/read', { args: { agentId: 'fx-alpha' } }))
.rejects.toThrow(/endpoint.*unavailable/)
})
})

View File

@@ -59,6 +59,7 @@ export class FakeApiClient implements IApiClient {
onModels: (payload: unknown) => Promise<RpcResponse<SessionModels>> = () => Promise.resolve(ok({
current: { provider: 'deepseek-official', model: 'deepseek-chat' },
routable: true,
groups: [],
failures: [],
}))

View File

@@ -28,7 +28,7 @@ describe('HTTP bridge abort', () => {
let carrierSignal: AbortSignal | undefined
const pending = bridge(request, response, {
fetch: async (input) => {
const fetchRequest = input as Request
const fetchRequest = input
carrierSignal = fetchRequest.signal
resolveStarted()
if (!fetchRequest.signal.aborted) {

View File

@@ -7,8 +7,9 @@ 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 { RpcId, type ClientRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
import type { HttpServerService, WebRoute, WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver'
import { API_PATH, apply, HOST_EVENTS_PATH, inject, MUX_EVENTS_PATH } from '../src/index.ts'
import { API_PATH, apply, HOST_EVENTS_PATH, inject, MUX_EVENTS_PATH, type HostConnectionHandle } from '../src/index.ts'
/** Structural httpServer fake recording both route registries. */
function fakeHttpServer(
@@ -17,6 +18,9 @@ function fakeHttpServer(
): Pick<HttpServerService, 'register' | 'registerUpgrade' | 'tapIndex' | 'port'> {
return {
register(route) {
if (routes.some(candidate => candidate.kind === route.kind && candidate.path === route.path)) {
throw new Error(`duplicate route ${route.path}`)
}
routes.push(route)
return () => { routes.splice(routes.indexOf(route), 1) }
},
@@ -36,15 +40,32 @@ function fakeRequest(headers: Record<string, string>, url = `${API_PATH}/session
return request
}
/** JSON POST carrying a complete client-request envelope. */
function fakePost(headers: Record<string, string>, url: string, body: unknown): IncomingMessage {
const request = Readable.from([Buffer.from(JSON.stringify(body))]) as unknown as IncomingMessage
Object.assign(request, { url, method: 'POST', headers: { 'content-type': 'application/json', ...headers } })
return request
}
/** Raw POST for malformed-body and media-type boundary cases. */
function fakeRawPost(headers: Record<string, string>, url: string, body: string): IncomingMessage {
const request = Readable.from([Buffer.from(body)]) as unknown as IncomingMessage
Object.assign(request, { url, method: 'POST', headers })
return request
}
/** Response recorder compatible with both the fence's short-circuit and the bridge. */
function fakeResponse(): { response: ServerResponse; state: { status?: number; body?: unknown } } {
const state: { status?: number; body?: unknown } = {}
const chunks: Buffer[] = []
const response = Object.assign(new EventEmitter(), {
writableEnded: false,
writeHead(value: number) { state.status = value; return this },
write() { return true },
write(value: string | Uint8Array) { chunks.push(Buffer.from(value)); return true },
end(this: { writableEnded: boolean }, value?: unknown) {
if (value !== undefined) state.body = value
if (typeof value === 'string' || value instanceof Uint8Array) chunks.push(Buffer.from(value))
else if (value !== undefined) throw new TypeError('fake response only accepts string or Uint8Array bodies')
if (chunks.length > 0) state.body = Buffer.concat(chunks).toString()
this.writableEnded = true
return this
},
@@ -173,6 +194,211 @@ describe('connection node half', () => {
expect(declared.state.status).toBe(404)
await dispose()
})
it('provides a disposable dedicated RPC channel without requiring apiProxy', async () => {
const ctx = new Context()
const routes: WebRoute[] = []
ctx.provide('httpServer', fakeHttpServer(routes, []) as HttpServerService)
const fiber = ctx.plugin({ inject: [...inject], apply })
await fiber.await()
expect(routes).toHaveLength(1)
expect(routes[0]).toMatchObject({ kind: 'prefix', path: API_PATH })
const connection = ctx.get('connection') as HostConnectionHandle
const calls: unknown[] = []
const remove = connection.rpc.handle('/rpc', async (endpoint, payload) => {
calls.push({ endpoint, payload })
return { ok: true, value: { accepted: true } }
}, { authority: 'trusted-host' })
const route = routes.find(candidate => candidate.path === '/rpc')
expect(route).toBeDefined()
const request: ClientRequest = {
type: 'client-request',
rpcId: RpcId('rpc-dedicated'),
method: 'goals/create',
payload: { args: { agentId: 'agent-1' } },
}
const result = fakeResponse()
await route!.handler(fakePost({ host: '127.0.0.1:3080' }, '/rpc/goals/create', request), result.response)
expect(result.state.status).toBe(200)
expect(JSON.parse(String(result.state.body))).toEqual({
type: 'server-response',
rpcId: 'rpc-dedicated',
result: { ok: true, value: { accepted: true } },
})
expect(calls).toEqual([{
endpoint: 'goals/create',
payload: { args: { agentId: 'agent-1' } },
}])
expect(() => connection.rpc.handle('/rpc', async () => ({ ok: true, value: null }), {
authority: 'trusted-host',
})).toThrow(/duplicate route/)
await remove()
expect(routes.map(candidate => candidate.path)).toEqual([API_PATH])
await fiber.dispose()
expect(routes).toHaveLength(0)
})
it('dispatches claimed /api endpoints before the API Proxy fallback and withdraws the claim', async () => {
const ctx = new Context()
const routes: WebRoute[] = []
ctx.provide('httpServer', fakeHttpServer(routes, []) as HttpServerService)
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
const calls: unknown[] = []
const remove = connection.rpc.intercept(
'/api',
endpoint => endpoint === 'goals/create',
async (endpoint, payload) => {
calls.push({ endpoint, payload })
return { ok: true, value: { accepted: true } }
},
{ authority: 'trusted-host' },
)
expect(() => connection.rpc.intercept(
'/api',
() => true,
async () => ({ ok: true, value: null }),
{ authority: 'trusted-host' },
)).toThrow('already has an interceptor')
expect(() => connection.rpc.intercept(
'/rpc' as '/api',
() => true,
async () => ({ ok: true, value: null }),
{ authority: 'trusted-host' },
)).toThrow('invalid shared RPC channel')
const route = routes.find(candidate => candidate.path === API_PATH)!
const request: ClientRequest = {
type: 'client-request',
rpcId: RpcId('rpc-shared'),
method: 'goals/create',
payload: { args: { agentId: 'agent-1' } },
}
const claimed = fakeResponse()
await route.handler(fakePost({ host: '127.0.0.1:3080' }, '/api/goals/create', request), claimed.response)
expect(JSON.parse(String(claimed.state.body))).toEqual({
type: 'server-response',
rpcId: 'rpc-shared',
result: { ok: true, value: { accepted: true } },
})
expect(calls).toEqual([{
endpoint: 'goals/create',
payload: { args: { agentId: 'agent-1' } },
}])
const denied = fakeResponse()
await route.handler(fakePost({ host: 'other.example' }, '/api/goals/create', request), denied.response)
expect(denied.state).toMatchObject({ status: 403, body: 'forbidden' })
expect(calls).toHaveLength(1)
const unclaimed = fakeResponse()
await route.handler(fakeRequest({ host: '127.0.0.1:3080' }, '/api/session.list'), unclaimed.response)
expect(unclaimed.state.status).toBe(404)
await remove()
const withdrawn = fakeResponse()
await route.handler(fakePost({ host: '127.0.0.1:3080' }, '/api/goals/create', request), withdrawn.response)
expect(withdrawn.state.status).toBe(404)
expect(calls).toHaveLength(1)
const removeLoopback = connection.rpc.intercept(
'/api',
endpoint => endpoint === 'goals/create',
async () => ({ ok: true, value: null }),
{ authority: 'loopback' },
)
const loopbackOnly = fakeResponse()
await route.handler(fakePost({ host: 'harness.example' }, '/api/goals/create', request), loopbackOnly.response)
expect(loopbackOnly.state.status).toBe(403)
await removeLoopback()
await fiber.dispose()
})
it('applies the configured trust fence and JSON envelope checks to generic channels', async () => {
const ctx = new Context()
const routes: WebRoute[] = []
ctx.provide('httpServer', fakeHttpServer(routes, []) as HttpServerService)
const fiber = ctx.plugin({ inject: [...inject], apply }, { trustedHosts: ['harness.example'] })
await fiber.await()
const connection = ctx.get('connection') as HostConnectionHandle
const remove = connection.rpc.handle('/rpc', async (endpoint) => {
if (endpoint === 'fail') throw new Error('handler broke')
return { ok: true, value: null }
}, {
authority: 'trusted-host',
})
const route = routes.find(candidate => candidate.path === '/rpc')!
const denied = fakeResponse()
await route.handler(fakePost({ host: 'other.example' }, '/rpc/goals/create', {}), denied.response)
expect(denied.state).toMatchObject({ status: 403, body: 'forbidden' })
const methodMismatch = fakeResponse()
await route.handler(fakePost({ host: 'harness.example' }, '/rpc/goals/create', {
type: 'client-request', rpcId: 'rpc-bad', method: 'other', payload: {},
}), methodMismatch.response)
expect(JSON.parse(String(methodMismatch.state.body))).toMatchObject({
rpcId: 'rpc-bad',
result: { ok: false, error: { code: 'bad-request' } },
})
for (const [request, status] of [
[fakeRequest({ host: 'harness.example' }, '/rpc/goals/create'), 404],
[fakePost({ host: 'harness.example' }, '/outside/goals/create', {}), 404],
[fakePost({ host: 'harness.example' }, '/rpc/goals//create', {}), 404],
[fakeRawPost({ host: 'harness.example' }, '/rpc/goals/create', '{}'), 415],
[fakeRawPost({ host: 'harness.example', 'content-type': 'text/plain' }, '/rpc/goals/create', '{}'), 415],
[fakeRawPost({ host: 'harness.example', 'content-type': 'application/json; charset=utf-8' }, '/rpc/goals/create', '{'), 400],
] as const) {
const response = fakeResponse()
await route.handler(request, response.response)
expect(response.state.status).toBe(status)
}
for (const [body, rpcId] of [
[{ rpcId: 'retained-id' }, 'retained-id'],
[{ rpcId: 42 }, 'invalid-request'],
[null, 'invalid-request'],
] as const) {
const response = fakeResponse()
await route.handler(fakePost({ host: 'harness.example' }, '/rpc/goals/create', body), response.response)
expect(JSON.parse(String(response.state.body))).toMatchObject({
rpcId,
result: { ok: false, error: { code: 'bad-request' } },
})
}
const failed = fakeResponse()
await route.handler(fakePost({ host: 'harness.example' }, '/rpc/fail', {
type: 'client-request', rpcId: 'rpc-fail', method: 'fail', payload: {},
}), failed.response)
expect(failed.state).toMatchObject({ status: 500, body: 'handler failure: Error: handler broke' })
expect(() => connection.rpc.handle('/api', async () => ({ ok: true, value: null }), {
authority: 'loopback',
})).toThrow('invalid or reserved RPC channel')
expect(() => connection.rpc.handle('api3', async () => ({ ok: true, value: null }), {
authority: 'loopback',
})).toThrow('invalid or reserved RPC channel')
const removeLoopback = connection.rpc.handle('/loopback', async () => ({ ok: true, value: null }), {
authority: 'loopback',
})
const loopbackRoute = routes.find(candidate => candidate.path === '/loopback')!
const publicResponse = fakeResponse()
await loopbackRoute.handler(fakePost({ host: 'harness.example' }, '/loopback/read', {
type: 'client-request', rpcId: 'rpc-public', method: 'read', payload: {},
}), publicResponse.response)
expect(publicResponse.state.status).toBe(403)
await removeLoopback()
await remove()
await fiber.dispose()
})
})
describe('connection node half over a real HTTP server', () => {