feat: typed eval output + assembly-reload/cancel handling

.changeset/assembly-reload-cancel.md

@@ -0,0 +1,11 @@
+---
+"@hotrepl/sdk": minor
+"@hotrepl/testing": minor
+---
+
+Handle assembly reloads and cancellation in the SDK. `Session.onAssemblyReload` surfaces hot-reload
+pushes that were previously dropped and invalidates the cached command catalog and descriptors so
+stale schemas are not reused. `Session.cancel(targetId)` cancels an active eval or subscription, and
+`watch()` now cancels its server subscription automatically when the iterator stops before `final`.
+`FakeRuntime` gains matching helpers (`onAssemblyReload`/`emitAssemblyReload`,
+`cancel`/`cancelled`).

.changeset/typed-eval-output.md

@@ -0,0 +1,13 @@
+---
+"@hotrepl/protocol": major
+"@hotrepl/sdk": major
+---
+
+Eval and subscription results now return properly typed output. `value` is emitted as native JSON
+instead of a JSON-encoded string, `valueType` carries the .NET type name, and a `truncated` /
+`truncatedBytes` pair signals when a result exceeds `maxResultLength` (in which case `value` is
+`null` rather than partial, invalid JSON).
+
+This is a breaking change for consumers that previously parsed `value` a second time.
+`Session.eval<T>()` and `Session.watch<T>()` now return the typed value directly and expose
+`truncated` / `truncatedBytes`.

docs/control-plane-protocol.md

@@ -76,10 +76,14 @@ JSON object.
 { "type": "eval", "id": "eval-1", "code": "1 + 1", "timeoutMs": 10000 }
 ```
 
-Success returns `eval_result`; failure returns `eval_error` with `error`:
+Success returns `eval_result`; failure returns `eval_error` with `error`. `value` is the result
+serialized to native JSON, so consumers receive properly typed output without a second parse;
+`valueType` carries the .NET type name. When the serialized value exceeds `maxResultLength`, `value`
+is `null`, `truncated` is `true`, and `truncatedBytes` reports the original byte size.
 
 ```json
-{ "type": "eval_result", "id": "eval-1", "hasValue": true, "value": "2", "durationMs": 3 }
+{ "type": "eval_result", "id": "eval-1", "hasValue": true, "value": 2, "valueType": "System.Int32", "truncated": false, "durationMs": 3 }
+{ "type": "eval_result", "id": "eval-2", "hasValue": true, "value": null, "valueType": "System.String", "truncated": true, "truncatedBytes": 250112, "durationMs": 8 }
 { "type": "eval_error", "id": "eval-1", "error": { "kind": "internal", "code": "runtimeException", "message": "...", "retryable": false } }
 ```
 
@@ -102,8 +106,10 @@ Success returns `eval_result`; failure returns `eval_error` with `error`:
 `onChange` (boolean, optional) triggers only when the expression value changes rather than every
 interval. `timeoutMs` (number, optional) bounds each tick.
 
-The server emits `subscribe_result` frames until `final: true`, or `subscribe_error` with `error`. A
-new client connection sends `session_evicted` to the old session and closes its subscriptions.
+The server emits `subscribe_result` frames until `final: true`, or `subscribe_error` with `error`.
+`subscribe_result` carries `value`/`truncated`/`truncatedBytes` with the same native-JSON semantics
+as `eval_result`. A new client connection sends `session_evicted` to the old session and closes its
+subscriptions.
 
 ## Typed commands
 
@@ -205,8 +211,8 @@ Entries include `id`, `kind`, optional `name` or `code`, `success`, `durationMs`
 
 ## Assembly Reload
 
-Sent by the server when a game assembly is hot-reloaded. Currently not handled by the SDK transport;
-clients may observe this as an unmatched server push.
+Sent by the server when a game assembly is hot-reloaded. The SDK surfaces this via
+`Session.onAssemblyReload` and invalidates its cached command catalog and descriptors.
 
 ```json
 {
@@ -220,8 +226,8 @@ clients may observe this as an unmatched server push.
 
 ### Cancel
 
-Cancel an active eval or subscription by its original request `id`. Not yet sent by the TypeScript
-SDK; available for custom transports.
+Cancel an active eval or subscription by its original request `id`. The SDK sends this via
+`Session.cancel(targetId)`, and automatically when a `watch()` iterator stops before `final`.
 
 ```json
 { "type": "cancel", "id": "cancel-1", "targetId": "watch-1" }

packages/protocol/schemas/eval_result.schema.json

@@ -21,6 +21,12 @@
     "valueType": {
       "type": "string"
     },
+    "truncated": {
+      "type": "boolean"
+    },
+    "truncatedBytes": {
+      "type": "number"
+    },
     "stdout": {
       "type": "string"
     },

packages/protocol/schemas/subscribe_result.schema.json

@@ -26,6 +26,12 @@
     "valueType": {
       "type": "string"
     },
+    "truncated": {
+      "type": "boolean"
+    },
+    "truncatedBytes": {
+      "type": "number"
+    },
     "durationMs": {
       "type": "number"
     },

packages/protocol/src/messages.ts

@@ -104,6 +104,8 @@ export const EvalResultMessageSchema = Type.Object(
     hasValue: Type.Boolean(),
     value: Type.Optional(Type.Unknown()),
     valueType: Type.Optional(Type.String()),
+    truncated: Type.Optional(Type.Boolean()),
+    truncatedBytes: Type.Optional(Type.Number()),
     stdout: Type.Optional(Type.String()),
     durationMs: Type.Number(),
   },
@@ -150,6 +152,8 @@ export const SubscribeResultMessageSchema = Type.Object(
     hasValue: Type.Boolean(),
     value: Type.Optional(Type.Unknown()),
     valueType: Type.Optional(Type.String()),
+    truncated: Type.Optional(Type.Boolean()),
+    truncatedBytes: Type.Optional(Type.Number()),
     durationMs: Type.Number(),
     final: Type.Boolean(),
   },
@@ -297,8 +301,8 @@ export const JournalQueryResultMessageSchema = Type.Object(
 );
 export type JournalQueryResultMessage = Static<typeof JournalQueryResultMessageSchema>;
 
-/** Sent by the server when a game assembly is hot-reloaded.
- *  Currently unhandled by the SDK transport (dropped silently). */
+/** Sent by the server when a game assembly is hot-reloaded. The SDK surfaces this via
+ *  Session.onAssemblyReload and invalidates its cached command catalog/descriptors. */
 export const AssemblyReloadMessageSchema = Type.Object(
   {
     type: Type.Literal(MESSAGE_TYPES.assemblyReload),
@@ -396,8 +400,8 @@ export const SubscribeMessageSchema = Type.Object(
 );
 export type SubscribeMessage = Static<typeof SubscribeMessageSchema>;
 
-/** Defined in C# CancelMessage; not yet used by the SDK RuntimeRequest.
- *  Cancels an active eval or subscription by its request id. */
+/** Cancels an active eval or subscription by its request id. Sent by the SDK via
+ *  Session.cancel(targetId), and automatically when a watch() iterator stops early. */
 export const CancelMessageSchema = Type.Object(
   {
     type: Type.Literal(MESSAGE_TYPES.cancel),

packages/sdk/src/session.ts

@@ -1,6 +1,7 @@
 import { MESSAGE_TYPES, PROTOCOL_VERSION } from "@hotrepl/protocol";
 import type {
   ArtifactRef,
+  AssemblyReloadMessage,
   CommandDescribeResultMessage,
   CommandDescriptor,
   CommandResultMessage,
@@ -45,6 +46,8 @@ export interface RuntimeTransport {
   watch(request: Extract<RuntimeRequest, { type: "subscribe" }>): AsyncIterable<WatchWireMessage>;
   readArtifact(ref: ArtifactRef): Promise<Uint8Array>;
   onSessionEvicted(listener: (event: SessionEvictedMessage) => void): () => void;
+  onAssemblyReload?(listener: (event: AssemblyReloadMessage) => void): () => void;
+  cancel?(targetId: string): void;
   close(): void;
 }
 
@@ -60,6 +63,8 @@ export interface EvalResponse<T = unknown> {
   hasValue: boolean;
   value?: T;
   valueType?: string;
+  truncated?: boolean;
+  truncatedBytes?: number;
   stdout?: string;
   durationMs: number;
 }
@@ -75,6 +80,8 @@ export interface WatchTick<T = unknown> {
   hasValue: boolean;
   value?: T;
   valueType?: string;
+  truncated?: boolean;
+  truncatedBytes?: number;
   final: boolean;
   durationMs: number;
 }
@@ -121,6 +128,7 @@ export class Session {
   private readonly descriptors: DescriptorCache = new Map();
   private catalog: CommandSummary[] | undefined;
   private readonly evictionListeners = new Set<(event: SessionEvictedMessage) => void>();
+  private readonly reloadListeners = new Set<(event: AssemblyReloadMessage) => void>();
   private sequence = 0;
   private evicted: SessionEvictedMessage | undefined;
   private closed = false;
@@ -132,13 +140,28 @@ export class Session {
       this.evicted = event;
       for (const listener of this.evictionListeners) listener(event);
     });
+    this.transport.onAssemblyReload?.((event) => {
+      // A hot-reloaded assembly can change registered commands and their schemas.
+      this.catalog = undefined;
+      this.descriptors.clear();
+      for (const listener of this.reloadListeners) listener(event);
+    });
   }
 
   onSessionEvicted(listener: (event: SessionEvictedMessage) => void): () => void {
     this.evictionListeners.add(listener);
     return () => this.evictionListeners.delete(listener);
   }
 
+  onAssemblyReload(listener: (event: AssemblyReloadMessage) => void): () => void {
+    this.reloadListeners.add(listener);
+    return () => this.reloadListeners.delete(listener);
+  }
+
+  cancel(targetId: string): void {
+    this.transport.cancel?.(targetId);
+  }
+
   async run<T = unknown>(
     name: string,
     args: unknown,
@@ -191,6 +214,8 @@ export class Session {
       durationMs: response.durationMs,
     };
     if (response.valueType !== undefined) result.valueType = response.valueType;
+    if (response.truncated !== undefined) result.truncated = response.truncated;
+    if (response.truncatedBytes !== undefined) result.truncatedBytes = response.truncatedBytes;
     if (response.stdout !== undefined) result.stdout = response.stdout;
     return result;
   }
@@ -239,6 +264,8 @@ export class Session {
       };
       if (event.hasValue && event.value !== undefined) tick.value = event.value as T;
       if (event.valueType !== undefined) tick.valueType = event.valueType;
+      if (event.truncated !== undefined) tick.truncated = event.truncated;
+      if (event.truncatedBytes !== undefined) tick.truncatedBytes = event.truncatedBytes;
       yield tick;
       if (event.final) return;
     }

packages/sdk/src/websocket-transport.ts

@@ -1,6 +1,7 @@
 import { MESSAGE_TYPES } from "@hotrepl/protocol";
 import type {
   ArtifactRef,
+  AssemblyReloadMessage,
   HandshakeMessage,
   ServerMessage,
   SessionEvictedMessage,
@@ -69,6 +70,8 @@ export class WebSocketTransport implements RuntimeTransport {
   private readonly pending = new Map<string, PendingRequest>();
   private readonly subscriptions = new Map<string, AsyncMessageQueue<WatchWireMessage>>();
   private readonly evictionListeners = new Set<(event: SessionEvictedMessage) => void>();
+  private readonly reloadListeners = new Set<(event: AssemblyReloadMessage) => void>();
+  private cancelSeq = 0;
   private handshakeMessage: HandshakeMessage | undefined;
   private evicted: SessionEvictedMessage | undefined;
 
@@ -110,14 +113,20 @@ export class WebSocketTransport implements RuntimeTransport {
     this.ensureAvailable();
     const queue = new AsyncMessageQueue<WatchWireMessage>();
     this.subscriptions.set(request.id, queue);
+    let final = false;
     try {
       this.socket.send(JSON.stringify(request));
       for await (const message of queue) {
         yield message;
-        if (message.final) return;
+        if (message.final) {
+          final = true;
+          return;
+        }
       }
     } finally {
       this.subscriptions.delete(request.id);
+      // If the consumer stopped early, tell the server to end the subscription.
+      if (!final) this.cancel(request.id);
     }
   }
 
@@ -142,6 +151,23 @@ export class WebSocketTransport implements RuntimeTransport {
     this.evictionListeners.add(listener);
     return () => this.evictionListeners.delete(listener);
   }
+
+  onAssemblyReload(listener: (event: AssemblyReloadMessage) => void): () => void {
+    this.reloadListeners.add(listener);
+    return () => this.reloadListeners.delete(listener);
+  }
+
+  cancel(targetId: string): void {
+    if (this.socket.readyState !== WebSocket.OPEN) return;
+    this.cancelSeq += 1;
+    try {
+      this.socket.send(
+        JSON.stringify({ type: MESSAGE_TYPES.cancel, id: `cancel-${this.cancelSeq}`, targetId }),
+      );
+    } catch {
+      // Cancellation is best-effort; a closing socket simply drops it.
+    }
+  }
   close(): void {
     this.socket.close();
   }
@@ -194,6 +220,11 @@ export class WebSocketTransport implements RuntimeTransport {
       return;
     }
 
+    if (message.type === MESSAGE_TYPES.assemblyReload) {
+      for (const listener of this.reloadListeners) listener(message);
+      return;
+    }
+
     if (message.type === MESSAGE_TYPES.error && message.id !== undefined) {
       const queue = this.subscriptions.get(message.id);
       if (queue !== undefined) {

packages/sdk/test/session.test.ts

@@ -56,6 +56,52 @@ describe("Session", () => {
     expect(runtime.requestCount("command_describe", "math.double")).toBe(0);
   });
 
+  test("eval returns native typed values and the truncation signal", async () => {
+    const runtime = new FakeRuntime();
+    runtime.setEvalHandler((code) =>
+      code === "big"
+        ? { hasValue: true, truncated: true, truncatedBytes: 4096 }
+        : { value: { x: 1, y: "z" }, valueType: "Anon" }
+    );
+    const session = await MockSession.create(runtime);
+
+    const obj = await session.eval<{ x: number; y: string }>("o");
+    expect(obj.value).toEqual({ x: 1, y: "z" });
+    expect(obj.truncated).toBeUndefined();
+
+    const big = await session.eval("big");
+    expect(big.hasValue).toBe(true);
+    expect(big.truncated).toBe(true);
+    expect(big.truncatedBytes).toBe(4096);
+    expect(big.value).toBeUndefined();
+  });
+
+  test("assembly_reload invalidates the command cache and notifies listeners", async () => {
+    const runtime = new FakeRuntime();
+    runtime.registerCommand(syncDescriptor, async () => ({ output: {} }));
+    const session = await MockSession.create(runtime);
+    await session.listCommands();
+
+    let reloaded = "";
+    session.onAssemblyReload((event) => {
+      reloaded = event.message;
+    });
+    runtime.emitAssemblyReload("rebuilt");
+
+    expect(reloaded).toBe("rebuilt");
+    await session.listCommands();
+    expect(runtime.requestCount("commands_list")).toBe(2);
+  });
+
+  test("cancel forwards the target id to the transport", async () => {
+    const runtime = new FakeRuntime();
+    const session = await MockSession.create(runtime);
+
+    session.cancel("eval-7");
+
+    expect(runtime.cancelled()).toContain("eval-7");
+  });
+
   test("run waits for job commands by polling job_status", async () => {
     const runtime = new FakeRuntime();
     runtime.registerCommand(

packages/sdk/test/websocket-transport.test.ts

@@ -23,7 +23,14 @@ function serveRuntime(runtime: FakeRuntime): { close: () => void; url: string }
         socket.data.closeEviction();
       },
       async message(socket, message) {
-        const request = JSON.parse(String(message)) as RuntimeRequest;
+        const parsed = JSON.parse(String(message)) as
+          | RuntimeRequest
+          | { type: "cancel"; targetId: string };
+        if (parsed.type === "cancel") {
+          runtime.cancel(parsed.targetId);
+          return;
+        }
+        const request = parsed;
         if (request.type === "subscribe") {
           for await (const event of runtime.watch(request)) {
             socket.send(JSON.stringify(event));