fix(rpc): preserve diagnostics and test native subscriptions

Author: dvcolomban

knip.jsonc

@@ -137,7 +137,7 @@
         "src/node/index.ts",
         "src/node/{auth,hub-internals}/index.ts",
         "src/recipes/interactive-auth.ts",
-        "src/rpc/{index,client,server}.ts",
+        "src/rpc/{index,client,server,shared-state}.ts",
         "src/rpc/dump/index.ts",
         "src/rpc/transports/{sse-client,sse-server,ws-bun,ws-deno,ws-client,ws-server}.ts",
         "src/types/index.ts",

packages/devframe/src/node/diagnostics.ts

@@ -1,4 +1,5 @@
 import { defineDiagnostics } from 'devframe/utils/nostics'
+import { sharedStateNotFound } from '../rpc/diagnostics'
 
 /**
  * DF00xx codes are allocated across packages (e.g. @devframes/json-render
@@ -20,6 +21,7 @@ export const diagnostics = defineDiagnostics({
     DF0012: {
       why: (p: { filepath: string }) => `Failed to parse storage file: ${p.filepath}, falling back to defaults.`,
     },
+    DF0013: sharedStateNotFound,
     DF0014: {
       why: (p: { name: string }) => `RPC function "${p.name}" has an invalid \`agent\` field: \`description\` must be a non-empty string.`,
       fix: 'Provide a short description (~1–3 sentences) explaining what the tool does and when agents should invoke it.',

packages/devframe/src/rpc/diagnostics.ts

@@ -1,11 +1,13 @@
 import { defineDiagnostics } from 'devframe/utils/nostics'
 
+export const sharedStateNotFound = {
+  why: (parameters: { key: string }) => `Shared state of "${parameters.key}" is not found, please provide an initial value for the first time`,
+} as const
+
 export const diagnostics = defineDiagnostics({
   docsBase: 'https://devfra.me/errors',
   codes: {
-    DF0013: {
-      why: (p: { key: string }) => `Shared state of "${p.key}" is not found, please provide an initial value for the first time`,
-    },
+    DF0013: sharedStateNotFound,
     DF0019: {
       why: (p: { name: string }) =>
         `RPC function "${p.name}" has \`agent\` set but \`jsonSerializable\` is \`false\`; MCP requires JSON-serializable data.`,

packages/devframe/src/rpc/shared-state.test.ts

@@ -3,13 +3,15 @@ import type { RpcClientEvents } from 'devframe/client'
 import type { DevframeRpcClientFunctions, DevframeRpcServerFunctions, RpcFunctionsHost } from 'devframe/types'
 import type { MessagePort } from 'node:worker_threads'
 import { MessageChannel } from 'node:worker_threads'
+import { createHostContext } from 'devframe/node'
 import { RpcFunctionsCollectorBase } from 'devframe/rpc'
 import { createRpcClient } from 'devframe/rpc/client'
 import { createRpcServer } from 'devframe/rpc/server'
 import { createRpcSharedStateClientHost, createRpcSharedStateServerHost } from 'devframe/rpc/shared-state'
 import { createEventEmitter } from 'devframe/utils/events'
 import { structuredCloneDeserialize, structuredCloneSerialize } from 'devframe/utils/structured-clone'
 import { expect, it } from 'vitest'
+import { createContextRpcServer } from '../node/rpc-core'
 
 /** JSON records also travel through transports that cannot clone native values. */
 function channelFor(port: MessagePort): ChannelOptions {
@@ -26,8 +28,21 @@ function channelFor(port: MessagePort): ChannelOptions {
   }
 }
 
-it('shares native state across independent channels with per-connection subscriptions', async () => {
-  expect.assertions(13)
+async function createStateServer(mode: string) {
+  if (mode === 'node context') {
+    const context = await createHostContext({
+      cwd: process.cwd(),
+      mode: 'dev',
+      host: {
+        mountStatic() {},
+        resolveOrigin: () => 'http://localhost',
+        getStorageDir: () => process.cwd(),
+      },
+    })
+    const { rpcGroup } = createContextRpcServer({ context, auth: false })
+    return { group: rpcGroup, sharedState: context.rpc.sharedState }
+  }
+
   const collector = new RpcFunctionsCollectorBase<DevframeRpcServerFunctions, undefined>(undefined)
   const group = createRpcServer<DevframeRpcClientFunctions, DevframeRpcServerFunctions>(collector.functions)
   const broadcast: RpcFunctionsHost['broadcast'] = async (options) => {
@@ -36,6 +51,12 @@ it('shares native state across independent channels with per-connection subscrip
       .map(client => client.$callRaw({ ...options, optional: true, event: true })))
   }
   const sharedState = createRpcSharedStateServerHost({ register: collector.register.bind(collector), broadcast })
+  return { group, sharedState }
+}
+
+it.each(['custom channels', 'node context'])('shares state across %s with per-connection subscriptions', async (mode) => {
+  expect.assertions(13)
+  const { group, sharedState } = await createStateServer(mode)
   const counter = await sharedState.get('counter', { initialValue: { count: 1 } })
   const channels = [new MessageChannel(), new MessageChannel()]
   const peers = channels.map((channel, index) => {

tests/__snapshots__/tsnapi/devframe/client.snapshot.d.ts

@@ -2,137 +2,9 @@
  * Generated by tsnapi — public API snapshot of `devframe/client`
  */
 // #region Interfaces
-export interface DevframeConnection {
-  connectionMeta: ConnectionMeta;
-  metaBaseUrl: string;
-  authToken?: string;
-}
-export interface DevframeRpcClient {
-  events: EventEmitter<RpcClientEvents>;
-  readonly isTrusted: boolean | null;
-  readonly status: DevframeConnectionStatus;
-  readonly connectionError: Error | null;
-  readonly transport: 'websocket' | 'sse' | 'static';
-  readonly connection: DevframeConnection;
-  readonly connectionMeta: ConnectionMeta;
-  ensureTrusted: (_?: number) => Promise<boolean>;
-  requestTrust: () => Promise<boolean>;
-  requestTrustWithToken: (_: string) => Promise<boolean>;
-  requestTrustWithCode: (_: string) => Promise<boolean>;
-  requestAuthCode: (_?: {
-    reissue?: boolean;
-  }) => Promise<void>;
-  call: DevframeRpcClientCall;
-  callEvent: DevframeRpcClientCallEvent;
-  callOptional: DevframeRpcClientCallOptional;
-  client: DevframeClientRpcHost;
-  sharedState: RpcSharedStateHost;
-  services: DevframeServicesClient;
-  streaming: RpcStreamingClientHost;
-  cacheManager: RpcCacheManager;
-  scope: {
-    <NS extends string>(_: NS): DevframeScopedClientContext<NS, SettingsForNamespace<NS>>;
-    (_?: null | ''): DevframeRpcClient;
-  };
-  close?: () => void;
-}
-export interface DevframeRpcClientMode {
-  readonly transport?: 'websocket' | 'sse' | 'static';
-  readonly isTrusted: boolean;
-  readonly status: DevframeConnectionStatus;
-  readonly connectionError: Error | null;
-  ensureTrusted: DevframeRpcClient['ensureTrusted'];
-  requestTrust: DevframeRpcClient['requestTrust'];
-  requestTrustWithToken: DevframeRpcClient['requestTrustWithToken'];
-  requestTrustWithCode: (_: string) => Promise<string | null>;
-  requestAuthCode: DevframeRpcClient['requestAuthCode'];
-  call: DevframeRpcClient['call'];
-  callEvent: DevframeRpcClient['callEvent'];
-  callOptional: DevframeRpcClient['callOptional'];
-  close?: () => void;
-}
-export interface DevframeRpcClientOptions extends SetupDevframeConnectionOptions {
-  authToken?: string;
-  otpParam?: string | false;
-  simpleAuth?: boolean;
-  transport?: 'auto' | 'websocket' | 'sse';
-  wsOptions?: Partial<WsRpcChannelOptions>;
-  sseOptions?: Partial<SseRpcChannelOptions>;
-  rpcOptions?: Partial<BirpcOptions<DevframeRpcServerFunctions, DevframeRpcClientFunctions, boolean>>;
-  cacheOptions?: boolean | Partial<RpcCacheOptions>;
-  webmcp?: boolean;
-  callTimeout?: number;
-}
-export interface DevframeRpcContext {
-  readonly rpc: DevframeRpcClient;
-}
-export interface DevframeScopedClientContext<NS extends string = string, Settings extends Record<string, any> = Record<string, any>> {
-  readonly namespace: NS;
-  readonly base: DevframeRpcClient;
-  rpc: DevframeScopedClientRpc<NS>;
-  settings: DevframeSettings<Settings>;
-  scope: DevframeRpcClient['scope'];
-}
-export interface DevframeScopedClientRpc<NS extends string = string> {
-  readonly namespace: NS;
-  register: (_: RpcFunctionDefinition<string, any, any, any, any, any, DevframeRpcContext>) => void;
-  call: {
-    <T extends keyof ScopedServerFunctions<NS> & string>(_: T, ..._: Parameters<ScopedRpcFn<DevframeRpcServerFunctions, NS, T>>): Promise<Awaited<ReturnType<ScopedRpcFn<DevframeRpcServerFunctions, NS, T>>>>;
-    <T extends keyof DevframeRpcServerFunctions & string>(_: T, ..._: Parameters<Extract<DevframeRpcServerFunctions[T], AnyRpcFn>>): Promise<Awaited<ReturnType<Extract<DevframeRpcServerFunctions[T], AnyRpcFn>>>>;
-    (_: string, ..._: any[]): Promise<any>;
-  };
-  callEvent: {
-    <T extends keyof ScopedServerFunctions<NS> & string>(_: T, ..._: Parameters<ScopedRpcFn<DevframeRpcServerFunctions, NS, T>>): void;
-    <T extends keyof DevframeRpcServerFunctions & string>(_: T, ..._: Parameters<Extract<DevframeRpcServerFunctions[T], AnyRpcFn>>): void;
-    (_: string, ..._: any[]): void;
-  };
-  callOptional: {
-    <T extends keyof ScopedServerFunctions<NS> & string>(_: T, ..._: Parameters<ScopedRpcFn<DevframeRpcServerFunctions, NS, T>>): Promise<Awaited<ReturnType<ScopedRpcFn<DevframeRpcServerFunctions, NS, T>>> | undefined>;
-    <T extends keyof DevframeRpcServerFunctions & string>(_: T, ..._: Parameters<Extract<DevframeRpcServerFunctions[T], AnyRpcFn>>): Promise<Awaited<ReturnType<Extract<DevframeRpcServerFunctions[T], AnyRpcFn>>> | undefined>;
-    (_: string, ..._: any[]): Promise<any>;
-  };
-  sharedState: {
-    <T extends keyof ScopedSharedStates<NS> & string>(_: T, _?: RpcSharedStateGetOptions<ScopedSharedStates<NS>[T]>): Promise<SharedState<ScopedSharedStates<NS>[T]>>;
-    <T extends Record<string, any> = Record<string, any>>(_: string, _?: RpcSharedStateGetOptions<T>): Promise<SharedState<T>>;
-  };
-  streaming: DevframeScopedClientStreamingHost;
-}
-export interface DevframeScopedClientStreamingHost {
-  subscribe: <T = unknown>(_: string, _: string, _?: StreamingSubscribeOptions) => StreamReader<T>;
-  upload: <T = unknown>(_: string, _: string) => StreamSink<T>;
-}
-export interface DevframeServiceClientHandle<NS extends string = string> extends DevframeServiceMeta {
-  readonly scope: NS;
-  readonly rpc: DevframeScopedClientRpc<NS>;
-}
-export interface DevframeServicesClient {
-  has: (_: string) => boolean;
-  get: <PKG extends string>(_: PKG) => DevframeServiceClientHandle<DevframeServiceScopeOf<PKG>> | undefined;
-  keys: () => string[];
-  state: () => Promise<SharedState<DevframeServicesState>>;
-}
 export interface RegisterWebMcpToolsOptions {
   modelContext?: WebMcpModelContext;
 }
-export interface RpcClientEvents {
-  'rpc:is-trusted:updated': (_: boolean) => void;
-  'connection:status': (_: DevframeConnectionStatus, _: DevframeConnectionStatus) => void;
-  'connection:error': (_: Error) => void;
-  'rpc:error': (_: Error, _: string) => void;
-}
-export interface RpcStreamingClientHost {
-  subscribe: <T = unknown>(_: string, _: string, _?: StreamingSubscribeOptions) => StreamReader<T>;
-  upload: <T = unknown>(_: string, _: string) => StreamSink<T>;
-}
-export interface SetupDevframeConnectionOptions {
-  connection?: DevframeConnection;
-  connectionMeta?: ConnectionMeta;
-  baseURL?: string | string[];
-  authToken?: string;
-}
-export interface StreamingSubscribeOptions {
-  highWaterMark?: number;
-}
 export interface WebMcpModelContext {
   registerTool: (_: WebMcpToolDescriptor, _?: {
     signal?: AbortSignal;
@@ -178,50 +50,51 @@ export interface WsUrlLocation {
 }
 // #endregion
 
-// #region Types
-export type DevframeClientRpcHost = RpcFunctionsCollector<DevframeRpcClientFunctions, DevframeRpcContext>;
-export type DevframeConnectionErrorKind = 'connection' | 'auth' | 'timeout';
-export type DevframeConnectionStatus = 'connecting' | 'connected' | 'unauthorized' | 'disconnected' | 'error';
-export type DevframeRpcClientCall = BirpcReturn<DevframeRpcServerFunctions, DevframeRpcClientFunctions>['$call'];
-export type DevframeRpcClientCallEvent = BirpcReturn<DevframeRpcServerFunctions, DevframeRpcClientFunctions>['$callEvent'];
-export type DevframeRpcClientCallOptional = BirpcReturn<DevframeRpcServerFunctions, DevframeRpcClientFunctions>['$callOptional'];
-// #endregion
-
-// #region Classes
-export declare class DevframeConnectionError extends Error {
-  name: string;
-  readonly kind: DevframeConnectionErrorKind;
-  constructor(_: DevframeConnectionErrorKind, _: string, _?: {
-    cause?: unknown;
-  });
-}
-// #endregion
-
 // #region Functions
 export declare function authenticateWithUrlOtp(_: Pick<DevframeRpcClient, 'isTrusted' | 'requestTrustWithCode'>, _?: {
   param?: string;
 }): Promise<boolean>;
 export declare function consumeOtpFromUrl(_?: string): string | undefined;
 export declare function createClientSettings<T extends Record<string, any> = Record<string, any>>(_: DevframeRpcClient, _: string): DevframeSettings<T>;
-export declare function createRpcStreamingClientHost(_: DevframeRpcClient): RpcStreamingClientHost;
-export declare function createScopedClientContext<NS extends string = string>(_: DevframeRpcClient, _: NS): DevframeScopedClientContext<NS>;
-export declare function getDevframeConnection(): DevframeConnection | undefined;
-export declare function getDevframeRpcClient(_?: DevframeRpcClientOptions): Promise<DevframeRpcClient>;
-export declare function isCallableStatus(_: DevframeConnectionStatus): boolean;
 export declare function readOtpFromUrl(_?: string): string | undefined;
-export declare function registerDevframeViewerOrigin(_: DevframeConnection, _?: any): Promise<boolean>;
 export declare function registerWebMcpTools<LocalFunctions, SetupContext>(_: RpcFunctionsCollector<LocalFunctions, SetupContext>, _?: RegisterWebMcpToolsOptions): () => void;
-export declare function resolveClientTransport(_: 'auto' | 'websocket' | 'sse', _: ConnectionMeta): 'websocket' | 'sse' | 'static';
 export declare function resolveSseUrl(_: ConnectionMeta['sse'], _: string, _: WsUrlLocation): string;
 export declare function resolveWebMcpModelContext(): WebMcpModelContext | undefined;
 export declare function resolveWsUrl(_: ConnectionMeta['websocket'], _: string, _: WsUrlLocation): string;
-export declare function setupDevframeConnection(_?: SetupDevframeConnectionOptions): Promise<DevframeConnection>;
 // #endregion
 
 // #region Variables
 export declare const connectDevframe: typeof getDevframeRpcClient;
 // #endregion
 
-// #region Referenced (internal)
-type AnyRpcFn = (..._: any[]) => any;
+// #region Other
+export { createRpcStreamingClientHost }
+export { createScopedClientContext }
+export { DevframeClientRpcHost }
+export { DevframeConnection }
+export { DevframeConnectionError }
+export { DevframeConnectionErrorKind }
+export { DevframeConnectionStatus }
+export { DevframeRpcClient }
+export { DevframeRpcClientCall }
+export { DevframeRpcClientCallEvent }
+export { DevframeRpcClientCallOptional }
+export { DevframeRpcClientMode }
+export { DevframeRpcClientOptions }
+export { DevframeRpcContext }
+export { DevframeScopedClientContext }
+export { DevframeScopedClientRpc }
+export { DevframeScopedClientStreamingHost }
+export { DevframeServiceClientHandle }
+export { DevframeServicesClient }
+export { getDevframeConnection }
+export { getDevframeRpcClient }
+export { isCallableStatus }
+export { registerDevframeViewerOrigin }
+export { resolveClientTransport }
+export { RpcClientEvents }
+export { RpcStreamingClientHost }
+export { setupDevframeConnection }
+export { SetupDevframeConnectionOptions }
+export { StreamingSubscribeOptions }
 // #endregion

tests/__snapshots__/tsnapi/devframe/internal.snapshot.d.ts

@@ -80,7 +80,7 @@ export declare const diagnostics: import("nostics").Diagnostics<{
     }) => string;
   };
   readonly DF0013: {
-    readonly why: (p: {
+    readonly why: (parameters: {
       key: string;
     }) => string;
   };
@@ -373,7 +373,7 @@ export declare const diagnostics: import("nostics").Diagnostics<{
   };
 }, readonly [(d: import("nostics").Diagnostic, { method }?: {
   method?: "log" | "warn" | "error";
-}) => void]>;
+}) => void], never>;
 // #endregion
 
 // #region Other

tests/__snapshots__/tsnapi/devframe/rpc.snapshot.js

@@ -1,41 +1,13 @@
 /**
  * Generated by tsnapi — public API snapshot of `devframe/rpc`
  */
-// #region Classes
-export class RpcCacheManager {
-  cacheMap
-  options
-  keySerializer
-  constructor(_) {}
-  updateOptions(_) {}
-  cached(_, _) {}
-  has(_, _) {}
-  apply(_, _) {}
-  validate(_) {}
-  clear(_) {}
-}
-export class RpcFunctionsCollectorBase {
-  context
-  definitions
-  functions
-  _onChanged
-  constructor(_) {}
-  register(_, _) {}
-  update(_, _) {}
-  onChanged(_) {}
-  async getHandler(_) {}
-  getSchema(_) {}
-  has(_) {}
-  get(_) {}
-  list() {}
-}
-// #endregion
-
 // #region Other
 export { createDefineWrapperWithContext }
 export { defineRpcFunction }
 export { getRpcHandler }
 export { getRpcResolvedSetupResult }
+export { RpcCacheManager }
+export { RpcFunctionsCollectorBase }
 export { strictJsonStringify }
 export { STRUCTURED_CLONE_PREFIX }
 export { validateDefinition }

tests/__snapshots__/tsnapi/devframe/rpc/shared-state.snapshot.d.ts

@@ -0,0 +1,10 @@
+/**
+ * Generated by tsnapi — public API snapshot of `devframe/rpc/shared-state`
+ */
+// #region Functions
+export declare function createRpcSharedStateClientHost<Context>(_: Pick<DevframeRpcClient, 'call' | 'callEvent' | 'events' | 'isTrusted'> & {
+  client: Pick<RpcFunctionsCollector<DevframeRpcClientFunctions, Context>, 'register'>;
+  connectionMeta: Pick<ConnectionMeta, 'backend'>;
+}): RpcSharedStateHost;
+export declare function createRpcSharedStateServerHost<Context>(_: Pick<RpcFunctionsCollector<DevframeRpcServerFunctions, Context>, 'register'> & Pick<RpcFunctionsHost, 'broadcast'>): RpcSharedStateHost;
+// #endregion

tests/__snapshots__/tsnapi/devframe/rpc/shared-state.snapshot.js

@@ -0,0 +1,10 @@
+/**
+ * Generated by tsnapi — public API snapshot of `devframe/rpc/shared-state`
+ */
+// #region Functions
+export function createRpcSharedStateServerHost(_) {}
+// #endregion
+
+// #region Other
+export { createRpcSharedStateClientHost }
+// #endregion