243f6629ef
The goal domain has been served by GoalService's @Remote namespace since it shipped; the API Proxy copy was a second implementation of the same six mutations. Remove the goals contract, schemas, route rows, IApiClient stub, host implementation, and the fixture's compatibility face, leaving ctx.remote.goals as the only path. The fixture's goal fold keeps its coverage through the Goal Remotes: its lifecycle case moves out of the unary-dispatch test, which no longer has goal rows to cover.
346 lines
18 KiB
TypeScript
346 lines
18 KiB
TypeScript
/**
|
|
* Client side of the fetch carrier. AbstractApiClient holds request correlation,
|
|
* envelope wrap/unwrap, zod parsing, and the payload-direct
|
|
* IApiClient domain methods (business code never mints). Platform differences ride two aspects:
|
|
* abstract doFetch (transport) + overridable onEnvelope (tap). ApiProxy (the impl face) is untouched.
|
|
*/
|
|
|
|
import type { z } from 'zod'
|
|
import { randomUUID } from '@deepseek-ai/dsh-util-crypto'
|
|
import type { RequestPayload, ResponseValue, RpcMethodMap } from '../api/rpc-map.ts'
|
|
import type { ClientRequest, RpcMessage, RpcResponse } from '../api/rpc.ts'
|
|
import { RpcId } from '../api/rpc.ts'
|
|
import type { Wire } from '../api/rpc.schema.ts'
|
|
import { serverResponseSchema } from '../api/rpc.schema.ts'
|
|
import {
|
|
hostCreateDirectoryValueSchema, hostDescribeValueSchema,
|
|
hostListDirectoryValueSchema, hostOpenPathValueSchema, hostPickDirectoryValueSchema,
|
|
} from '../api/host.schema.ts'
|
|
import { skillListValueSchema } from '../api/skills.schema.ts'
|
|
import {
|
|
agentPresetCopyValueSchema, agentPresetListValueSchema, agentPresetOpenDocumentValueSchema,
|
|
agentPresetReadValueSchema, agentPresetRemoveValueSchema, agentPresetSelectValueSchema,
|
|
} from '../api/agent-presets.schema.ts'
|
|
import {
|
|
settingsDescribeValueSchema, settingsMutateValueSchema, settingsOpenDocumentValueSchema,
|
|
settingsReplaceValueSchema, settingsUpdateValueSchema,
|
|
} from '../api/settings.schema.ts'
|
|
import {
|
|
credentialsDescribeValueSchema, credentialsSetValueSchema, credentialsUnsetValueSchema,
|
|
} from '../api/credentials.schema.ts'
|
|
import { llmDiscoverModelsValueSchema, llmModelsValueSchema, llmProvidersValueSchema } from '../api/llm.schema.ts'
|
|
import {
|
|
subagentInterruptValueSchema,
|
|
subagentListValueSchema,
|
|
subagentPromptValueSchema,
|
|
} from '../api/subagents.schema.ts'
|
|
|
|
/**
|
|
* Client consumption face of the contract (shape a): same domain tree as ApiProxy, but unary
|
|
* methods take the business payload directly — the carrier mints the rpcId and wraps the
|
|
* envelope. Business code needing the call's rpcId reads it from the RpcResponse echo.
|
|
* Unary methods accept an optional external AbortSignal as the last parameter.
|
|
* Bounded calls merge it with the instance timeout via AbortSignal.any; user-paced calls
|
|
* carry only that external signal. In both cases the signal rides beside the request, never
|
|
* on the wire, like the stream signatures.
|
|
* Relationship: ApiProxy is the narrow-form signature contract the impl side implements;
|
|
* IApiClient is the payload-direct view clients consume; AbstractApiClient bridges the two.
|
|
* Derived per method key from RpcMethodMap so a map row addition updates this mechanically.
|
|
*/
|
|
export interface IApiClient {
|
|
subagents: {
|
|
list(payload: RequestPayload<'subagent.list'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'subagent.list'>>>
|
|
prompt(payload: RequestPayload<'subagent.prompt'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'subagent.prompt'>>>
|
|
interrupt(payload: RequestPayload<'subagent.interrupt'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'subagent.interrupt'>>>
|
|
}
|
|
host: {
|
|
describe(payload: RequestPayload<'host.describe'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'host.describe'>>>
|
|
pickDirectory(payload: RequestPayload<'host.pickDirectory'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'host.pickDirectory'>>>
|
|
listDirectory(payload: RequestPayload<'host.listDirectory'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'host.listDirectory'>>>
|
|
createDirectory(payload: RequestPayload<'host.createDirectory'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'host.createDirectory'>>>
|
|
openPath(payload: RequestPayload<'host.openPath'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'host.openPath'>>>
|
|
}
|
|
skills: {
|
|
list(payload: RequestPayload<'skill.list'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'skill.list'>>>
|
|
}
|
|
agentPresets: {
|
|
list(payload: RequestPayload<'agentPreset.list'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'agentPreset.list'>>>
|
|
select(payload: RequestPayload<'agentPreset.select'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'agentPreset.select'>>>
|
|
read(payload: RequestPayload<'agentPreset.read'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'agentPreset.read'>>>
|
|
copy(payload: RequestPayload<'agentPreset.copy'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'agentPreset.copy'>>>
|
|
openDocument(payload: RequestPayload<'agentPreset.openDocument'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'agentPreset.openDocument'>>>
|
|
remove(payload: RequestPayload<'agentPreset.remove'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'agentPreset.remove'>>>
|
|
}
|
|
settings: {
|
|
describe(payload: RequestPayload<'settings.describe'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'settings.describe'>>>
|
|
openDocument(payload: RequestPayload<'settings.openDocument'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'settings.openDocument'>>>
|
|
update(payload: RequestPayload<'settings.update'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'settings.update'>>>
|
|
replace(payload: RequestPayload<'settings.replace'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'settings.replace'>>>
|
|
mutate(payload: RequestPayload<'settings.mutate'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'settings.mutate'>>>
|
|
}
|
|
credentials: {
|
|
describe(payload: RequestPayload<'credentials.describe'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'credentials.describe'>>>
|
|
set(payload: RequestPayload<'credentials.set'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'credentials.set'>>>
|
|
unset(payload: RequestPayload<'credentials.unset'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'credentials.unset'>>>
|
|
}
|
|
llm: {
|
|
providers(payload: RequestPayload<'llm.providers'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'llm.providers'>>>
|
|
models(payload: RequestPayload<'llm.models'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'llm.models'>>>
|
|
discoverModels(payload: RequestPayload<'llm.discoverModels'>, signal?: AbortSignal): Promise<RpcResponse<ResponseValue<'llm.discoverModels'>>>
|
|
}
|
|
}
|
|
|
|
/**
|
|
* S→C second-level parse table: value schema by method (the response-path
|
|
* mirror of the handler's request table; key coverage compiler-enforced against RpcMethodMap).
|
|
*/
|
|
const UNARY_VALUE_SCHEMAS: { [K in keyof RpcMethodMap]: z.ZodType<Wire<ResponseValue<K>>> } = {
|
|
'subagent.list': subagentListValueSchema,
|
|
'subagent.prompt': subagentPromptValueSchema,
|
|
'subagent.interrupt': subagentInterruptValueSchema,
|
|
'host.describe': hostDescribeValueSchema,
|
|
'host.pickDirectory': hostPickDirectoryValueSchema,
|
|
'host.listDirectory': hostListDirectoryValueSchema,
|
|
'host.createDirectory': hostCreateDirectoryValueSchema,
|
|
'host.openPath': hostOpenPathValueSchema,
|
|
'skill.list': skillListValueSchema,
|
|
'agentPreset.list': agentPresetListValueSchema,
|
|
'agentPreset.select': agentPresetSelectValueSchema,
|
|
'agentPreset.read': agentPresetReadValueSchema,
|
|
'agentPreset.copy': agentPresetCopyValueSchema,
|
|
'agentPreset.openDocument': agentPresetOpenDocumentValueSchema,
|
|
'agentPreset.remove': agentPresetRemoveValueSchema,
|
|
'settings.describe': settingsDescribeValueSchema,
|
|
'settings.openDocument': settingsOpenDocumentValueSchema,
|
|
'settings.update': settingsUpdateValueSchema,
|
|
'settings.replace': settingsReplaceValueSchema,
|
|
'settings.mutate': settingsMutateValueSchema,
|
|
'credentials.describe': credentialsDescribeValueSchema,
|
|
'credentials.set': credentialsSetValueSchema,
|
|
'credentials.unset': credentialsUnsetValueSchema,
|
|
'llm.providers': llmProvidersValueSchema,
|
|
'llm.models': llmModelsValueSchema,
|
|
'llm.discoverModels': llmDiscoverModelsValueSchema,
|
|
}
|
|
|
|
/** Default timeout for bounded unary calls (rpc-compare 2026-07-19: a hung host must not leave callers pending forever). */
|
|
const DEFAULT_TIMEOUT_MS = 30_000
|
|
|
|
/** Whether a unary call uses the transport health deadline or only caller/connection cancellation. */
|
|
type UnaryTimeoutPolicy = 'default' | 'caller-signal-only'
|
|
|
|
/** URL base for in-process handler injection (fake authority, opencode precedent). */
|
|
const INTERNAL_BASE = 'http://dsh.internal'
|
|
|
|
/**
|
|
* Abstract fetch-carrier client. Subclasses supply the transport (doFetch) and may refine the
|
|
* per-message tap (onEnvelope) — platform aspects stay in subclasses, protocol invariants stay
|
|
* here. Envelope observation is a first-class aspect of this data middle layer: the instance
|
|
* owns a microtask-batched buffer (frame storms must not cost one consumer update per frame),
|
|
* and observers subscribe via subscribeEnvelopes. The isomorphic point survives: an in-process
|
|
* subclass whose doFetch is toFetchHandler(api).fetch never touches the network.
|
|
*/
|
|
export abstract class AbstractApiClient implements IApiClient {
|
|
/** Instance-owned observation buffer (module-level state would leak across instances/tests). */
|
|
private envelopeBatch: RpcMessage[] = []
|
|
private flushScheduled = false
|
|
private readonly envelopeListeners = new Set<(batch: readonly RpcMessage[]) => void>()
|
|
|
|
/** @param timeoutMs - timeout for bounded unary calls; user-paced calls do not use it. */
|
|
constructor(protected readonly timeoutMs: number = DEFAULT_TIMEOUT_MS) {}
|
|
|
|
/** Transport aspect: browser fetch, injected handler.fetch, IPC bridge, ... */
|
|
protected abstract doFetch(input: URL, init?: RequestInit): Promise<Response>
|
|
|
|
/**
|
|
* Subscribe to batched envelope observation (diagnostics/logging consumers).
|
|
* Batches follow microtask boundaries; a listener throw is isolated (observation
|
|
* must never break the carrier).
|
|
* @param listener - receives each flushed batch in arrival order.
|
|
* @returns unsubscribe function.
|
|
*/
|
|
subscribeEnvelopes(listener: (batch: readonly RpcMessage[]) => void): () => void {
|
|
this.envelopeListeners.add(listener)
|
|
return () => {
|
|
this.envelopeListeners.delete(listener)
|
|
}
|
|
}
|
|
|
|
/** Per-message tap: feeds the instance buffer. Subclasses may override to observe unbatched (call super to keep batching). */
|
|
protected onEnvelope(message: RpcMessage): void {
|
|
if (this.envelopeListeners.size === 0) return
|
|
this.envelopeBatch.push(message)
|
|
if (this.flushScheduled) return
|
|
this.flushScheduled = true
|
|
queueMicrotask(() => {
|
|
this.flushScheduled = false
|
|
// Never empty here: a flush is only ever scheduled by the push above,
|
|
// and this callback is the sole drain point.
|
|
const batch = this.envelopeBatch
|
|
this.envelopeBatch = []
|
|
for (const notify of this.envelopeListeners) {
|
|
try {
|
|
notify(batch)
|
|
} catch (error) {
|
|
console.error('[apiproxy] envelope listener threw:', error)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
/** Browser = same-origin (a fake authority would fail DNS on real requests); no-location env (Node) = fake authority. */
|
|
protected resolveBase(): string {
|
|
const loc = (globalThis as { location?: { origin?: string } }).location
|
|
return loc?.origin !== undefined && loc.origin !== 'null' ? loc.origin : INTERNAL_BASE
|
|
}
|
|
|
|
protected mintRpcId(): RpcId {
|
|
// Not crypto.randomUUID: browsers withhold it outside secure contexts,
|
|
// and this base also mints on pages served over plain HTTP.
|
|
return RpcId(randomUUID())
|
|
}
|
|
|
|
/**
|
|
* Shared POST leg of unary calls: JSON body,
|
|
* optional default timeout merged with the caller's external signal, non-2xx → transport throw.
|
|
*/
|
|
private async postJson(
|
|
path: string,
|
|
body: ClientRequest,
|
|
signal: AbortSignal | undefined,
|
|
timeoutPolicy: UnaryTimeoutPolicy = 'default',
|
|
): Promise<Response> {
|
|
const requestSignal = timeoutPolicy === 'default'
|
|
? signal === undefined
|
|
? AbortSignal.timeout(this.timeoutMs)
|
|
: AbortSignal.any([AbortSignal.timeout(this.timeoutMs), signal])
|
|
: signal
|
|
const response = await this.doFetch(new URL(path, this.resolveBase()), {
|
|
method: 'POST',
|
|
headers: { 'content-type': 'application/json' },
|
|
body: JSON.stringify(body),
|
|
...requestSignal === undefined ? {} : { signal: requestSignal },
|
|
})
|
|
if (!response.ok) throw new Error(`transport failure for ${path}: HTTP ${response.status}`)
|
|
return response
|
|
}
|
|
|
|
/**
|
|
* Unary protocol path: mint → tap → POST full form → envelope parse → verify
|
|
* echo → value parse → tap → narrow. Virtual so a fake carrier (fixture) can
|
|
* override transport at this layer.
|
|
*/
|
|
protected async callUnary<K extends keyof RpcMethodMap>(
|
|
method: K,
|
|
payload: RequestPayload<K>,
|
|
signal?: AbortSignal,
|
|
timeoutPolicy: UnaryTimeoutPolicy = 'default',
|
|
): Promise<RpcResponse<ResponseValue<K>>> {
|
|
const message: ClientRequest = { type: 'client-request', rpcId: this.mintRpcId(), method, payload }
|
|
this.onEnvelope(message)
|
|
const response = await this.postJson(`/api/${method}`, message, signal, timeoutPolicy)
|
|
const full = serverResponseSchema.parse(await response.json())
|
|
this.onEnvelope(full)
|
|
if (full.rpcId !== message.rpcId) throw new Error(`rpcId mismatch for ${method}: sent ${message.rpcId}, got ${full.rpcId}`)
|
|
if (!full.result.ok) return { rpcId: full.rpcId, result: full.result }
|
|
// Second-level S→C parse: the ok value must match the method's Value schema (mirror of the
|
|
// handler's request-payload parse). The cast collapses the Wire<> widening, same as the handler side.
|
|
const value = UNARY_VALUE_SCHEMAS[method].parse(full.result.value) as ResponseValue<K>
|
|
return { rpcId: full.rpcId, result: { ok: true, value } }
|
|
}
|
|
|
|
// ---- IApiClient API (arrow properties so destructured/passed references stay bound) ----
|
|
|
|
readonly subagents: IApiClient['subagents'] = {
|
|
list: (payload, signal) => this.callUnary('subagent.list', payload, signal),
|
|
prompt: (payload, signal) => this.callUnary('subagent.prompt', payload, signal),
|
|
interrupt: (payload, signal) => this.callUnary('subagent.interrupt', payload, signal),
|
|
}
|
|
|
|
readonly host: IApiClient['host'] = {
|
|
describe: (payload, signal) => this.callUnary('host.describe', payload, signal),
|
|
// A native system dialog is user-paced and may legitimately stay open
|
|
// longer than the normal unary deadline. Caller/connection aborts remain.
|
|
pickDirectory: (payload, signal) => this.callUnary(
|
|
'host.pickDirectory', payload, signal, 'caller-signal-only',
|
|
),
|
|
listDirectory: (payload, signal) => this.callUnary('host.listDirectory', payload, signal),
|
|
createDirectory: (payload, signal) => this.callUnary('host.createDirectory', payload, signal),
|
|
openPath: (payload, signal) => this.callUnary('host.openPath', payload, signal),
|
|
}
|
|
|
|
readonly skills: IApiClient['skills'] = {
|
|
list: (payload, signal) => this.callUnary('skill.list', payload, signal),
|
|
}
|
|
|
|
// Annotated like every sibling, and load-bearing rather than cosmetic:
|
|
// inferring this member inlines `AgentPresetEntry` into the emitted
|
|
// declaration by the specifier TS picks — the host `index.ts` — which drags
|
|
// the whole gateway, and with it the host `Context` merges, into every
|
|
// Client program that imports this carrier.
|
|
readonly agentPresets: IApiClient['agentPresets'] = {
|
|
list: (payload, signal) => this.callUnary('agentPreset.list', payload, signal),
|
|
select: (payload, signal) => this.callUnary('agentPreset.select', payload, signal),
|
|
read: (payload, signal) => this.callUnary('agentPreset.read', payload, signal),
|
|
copy: (payload, signal) => this.callUnary('agentPreset.copy', payload, signal),
|
|
openDocument: (payload, signal) => this.callUnary('agentPreset.openDocument', payload, signal),
|
|
remove: (payload, signal) => this.callUnary('agentPreset.remove', payload, signal),
|
|
}
|
|
|
|
readonly settings: IApiClient['settings'] = {
|
|
describe: (payload, signal) => this.callUnary('settings.describe', payload, signal),
|
|
openDocument: (payload, signal) => this.callUnary('settings.openDocument', payload, signal),
|
|
update: (payload, signal) => this.callUnary('settings.update', payload, signal),
|
|
replace: (payload, signal) => this.callUnary('settings.replace', payload, signal),
|
|
mutate: (payload, signal) => this.callUnary('settings.mutate', payload, signal),
|
|
}
|
|
|
|
readonly credentials: IApiClient['credentials'] = {
|
|
describe: (payload, signal) => this.callUnary('credentials.describe', payload, signal),
|
|
set: (payload, signal) => this.callUnary('credentials.set', payload, signal),
|
|
unset: (payload, signal) => this.callUnary('credentials.unset', payload, signal),
|
|
}
|
|
|
|
readonly llm: IApiClient['llm'] = {
|
|
providers: (payload, signal) => this.callUnary('llm.providers', payload, signal),
|
|
models: (payload, signal) => this.callUnary('llm.models', payload, signal),
|
|
discoverModels: (payload, signal) => this.callUnary('llm.discoverModels', payload, signal),
|
|
}
|
|
|
|
}
|
|
|
|
/**
|
|
* In-process client over an injected fetch-shaped handler (the isomorphic point:
|
|
* `new InProcessApiClient(toFetchHandler(api))` never touches the network). Lives here because
|
|
* in-process injection is this package's own capability (handler and client are both local).
|
|
*/
|
|
export class InProcessApiClient extends AbstractApiClient {
|
|
constructor(private readonly handler: { fetch: typeof fetch }, timeoutMs?: number) {
|
|
super(timeoutMs)
|
|
}
|
|
|
|
/**
|
|
* Faithful to real fetch: reject on signal abort even when the in-process
|
|
* handler ignores the signal (a hung impl must not defeat timeout/cancel).
|
|
*/
|
|
protected doFetch(input: URL, init?: RequestInit): Promise<Response> {
|
|
const signal = init?.signal ?? undefined
|
|
if (signal === undefined) return this.handler.fetch(input, init)
|
|
if (signal.aborted) return Promise.reject(abortError(signal))
|
|
return new Promise((resolve, reject) => {
|
|
const onAbort = (): void => { reject(abortError(signal)) }
|
|
signal.addEventListener('abort', onAbort, { once: true })
|
|
this.handler.fetch(input, init)
|
|
.then(resolve, reject)
|
|
.finally(() => { signal.removeEventListener('abort', onAbort) })
|
|
})
|
|
}
|
|
}
|
|
|
|
/** Mirror fetch's abort rejection: the signal's reason when present, else a DOMException-style AbortError. */
|
|
function abortError(signal: AbortSignal): Error {
|
|
const reason: unknown = signal.reason
|
|
if (reason instanceof Error) return reason
|
|
if (typeof reason === 'string') return new Error(reason)
|
|
return new Error('This operation was aborted')
|
|
}
|