From 6c31ae198e00cf4a43a5316f3b75eac761202514 Mon Sep 17 00:00:00 2001 From: Johann Glock <11704293+glockyco@users.noreply.github.com> Date: Thu, 4 Jun 2026 20:05:59 +0200 Subject: [PATCH 1/2] feat(protocol): emit native typed eval and subscription values Eval and subscription results serialized the value to JSON and then sent it as a JSON-encoded string, so every consumer had to parse it a second time and the SDK's typed eval/watch actually returned a string rather than T. Emit the value as native JSON instead, carry the .NET type name in valueType, and signal oversized results with truncated/truncatedBytes (value null) rather than partial, invalid JSON. The SDK now returns the typed value directly and exposes the truncation fields, and the fake runtime mirrors the truncation contract so tests exercise it. BREAKING CHANGE: eval_result/subscribe_result value is native JSON instead of a JSON-encoded string; consumers that double-parsed value must stop. --- .changeset/typed-eval-output.md | 13 +++++ docs/control-plane-protocol.md | 14 +++-- .../protocol/schemas/eval_result.schema.json | 6 ++ .../schemas/subscribe_result.schema.json | 6 ++ packages/protocol/src/messages.ts | 4 ++ packages/sdk/src/session.ts | 8 +++ packages/sdk/test/session.test.ts | 20 +++++++ packages/testing/src/fake-runtime.ts | 16 ++++- src/HotRepl.Core/ReplEngine.cs | 18 +++--- .../Serialization/JsonResultSerializer.cs | 51 ++++++++++++++++ .../Subscriptions/SubscriptionManager.cs | 9 ++- .../Messages/Outbound/EvalResultMessage.cs | 6 ++ .../Outbound/SubscribeResultMessage.cs | 6 ++ .../Unit/ResultSerializerTests.cs | 58 ++++++++++++++++++- 14 files changed, 211 insertions(+), 24 deletions(-) create mode 100644 .changeset/typed-eval-output.md diff --git a/.changeset/typed-eval-output.md b/.changeset/typed-eval-output.md new file mode 100644 index 0000000..f81e0fb --- /dev/null +++ b/.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()` and `Session.watch()` now return the typed value directly and expose +`truncated` / `truncatedBytes`. diff --git a/docs/control-plane-protocol.md b/docs/control-plane-protocol.md index 644c6db..37640ac 100644 --- a/docs/control-plane-protocol.md +++ b/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 diff --git a/packages/protocol/schemas/eval_result.schema.json b/packages/protocol/schemas/eval_result.schema.json index 94e9178..cfefa4d 100644 --- a/packages/protocol/schemas/eval_result.schema.json +++ b/packages/protocol/schemas/eval_result.schema.json @@ -21,6 +21,12 @@ "valueType": { "type": "string" }, + "truncated": { + "type": "boolean" + }, + "truncatedBytes": { + "type": "number" + }, "stdout": { "type": "string" }, diff --git a/packages/protocol/schemas/subscribe_result.schema.json b/packages/protocol/schemas/subscribe_result.schema.json index b873dea..d6e493c 100644 --- a/packages/protocol/schemas/subscribe_result.schema.json +++ b/packages/protocol/schemas/subscribe_result.schema.json @@ -26,6 +26,12 @@ "valueType": { "type": "string" }, + "truncated": { + "type": "boolean" + }, + "truncatedBytes": { + "type": "number" + }, "durationMs": { "type": "number" }, diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 08484c0..1a6c297 100644 --- a/packages/protocol/src/messages.ts +++ b/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(), }, diff --git a/packages/sdk/src/session.ts b/packages/sdk/src/session.ts index 859f096..e4a28ad 100644 --- a/packages/sdk/src/session.ts +++ b/packages/sdk/src/session.ts @@ -60,6 +60,8 @@ export interface EvalResponse { hasValue: boolean; value?: T; valueType?: string; + truncated?: boolean; + truncatedBytes?: number; stdout?: string; durationMs: number; } @@ -75,6 +77,8 @@ export interface WatchTick { hasValue: boolean; value?: T; valueType?: string; + truncated?: boolean; + truncatedBytes?: number; final: boolean; durationMs: number; } @@ -191,6 +195,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 +245,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; } diff --git a/packages/sdk/test/session.test.ts b/packages/sdk/test/session.test.ts index 3106f74..43c2e6f 100644 --- a/packages/sdk/test/session.test.ts +++ b/packages/sdk/test/session.test.ts @@ -56,6 +56,26 @@ 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("run waits for job commands by polling job_status", async () => { const runtime = new FakeRuntime(); runtime.registerCommand( diff --git a/packages/testing/src/fake-runtime.ts b/packages/testing/src/fake-runtime.ts index f95125a..2ccac9d 100644 --- a/packages/testing/src/fake-runtime.ts +++ b/packages/testing/src/fake-runtime.ts @@ -47,6 +47,14 @@ type WatchEvent = | { hasValue?: boolean; value?: unknown; valueType?: string; final: boolean } | { error: HotReplErrorEnvelope; final: boolean }; +type FakeEvalResult = { + value?: unknown; + valueType?: string; + hasValue?: boolean; + truncated?: boolean; + truncatedBytes?: number; +}; + export interface FakeRuntimeOptions { protocolVersion?: number; supportsCompletion?: boolean; @@ -66,7 +74,7 @@ export class FakeRuntime implements RuntimeTransport { private readonly journalEntries: JournalEntry[] = []; private readonly counters = new Map(); private readonly evictionListeners = new Set<(event: SessionEvictedMessage) => void>(); - private evalHandler: (code: string) => { value?: unknown; valueType?: string } = () => ({}); + private evalHandler: (code: string) => FakeEvalResult = () => ({}); private activeRequests = 0; private nextJob = 0; private isClosed = false; @@ -116,7 +124,7 @@ export class FakeRuntime implements RuntimeTransport { }); } - setEvalHandler(handler: (code: string) => { value?: unknown; valueType?: string }): void { + setEvalHandler(handler: (code: string) => FakeEvalResult): void { this.evalHandler = handler; } @@ -271,11 +279,13 @@ export class FakeRuntime implements RuntimeTransport { const response: EvalResultMessage = { type: MESSAGE_TYPES.evalResult, id: request.id, - hasValue: value !== undefined, + hasValue: result.hasValue ?? value !== undefined, value, durationMs: 0, }; if (result.valueType !== undefined) response.valueType = result.valueType; + if (result.truncated !== undefined) response.truncated = result.truncated; + if (result.truncatedBytes !== undefined) response.truncatedBytes = result.truncatedBytes; return response; } catch (caught) { const envelope = toEnvelope(caught); diff --git a/src/HotRepl.Core/ReplEngine.cs b/src/HotRepl.Core/ReplEngine.cs index f6faec1..0f3d0ed 100644 --- a/src/HotRepl.Core/ReplEngine.cs +++ b/src/HotRepl.Core/ReplEngine.cs @@ -18,7 +18,6 @@ using HotRepl.Serialization; using HotRepl.Server; using HotRepl.Subscriptions; -using Newtonsoft.Json.Linq; namespace HotRepl; @@ -746,21 +745,18 @@ private void SendEvalOutcome(string id, Guid connectionId, EvalOutcome outcome) string json; if (outcome.Success) { - string? serialized = null; - if (outcome.HasValue && outcome.Value != null) - { - serialized = JsonResultSerializer.Serialize(outcome.Value, _host.Config); - serialized = JsonResultSerializer.Truncate( - serialized, - _host.Config.MaxResultLength - ); - } + var wire = + outcome.HasValue && outcome.Value != null + ? JsonResultSerializer.ToWireValue(outcome.Value, _host.Config) + : default; json = Serialize( new EvalResultMessage { Id = id, HasValue = outcome.HasValue, - Value = serialized == null ? null : JToken.FromObject(serialized), + Value = wire.Value, + Truncated = wire.Truncated, + TruncatedBytes = wire.ByteCount, ValueType = outcome.ValueType, Stdout = string.IsNullOrEmpty(outcome.Stdout) ? null : outcome.Stdout, DurationMs = outcome.DurationMs, diff --git a/src/HotRepl.Core/Serialization/JsonResultSerializer.cs b/src/HotRepl.Core/Serialization/JsonResultSerializer.cs index bf75718..e1f7ca0 100644 --- a/src/HotRepl.Core/Serialization/JsonResultSerializer.cs +++ b/src/HotRepl.Core/Serialization/JsonResultSerializer.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.Text; using Newtonsoft.Json; +using Newtonsoft.Json.Linq; namespace HotRepl.Serialization; @@ -50,6 +51,56 @@ public static string Serialize(object? value, ReplConfig config) } } + /// A typed wire value plus a truncation signal for oversized results. + internal readonly struct WireValue + { + public WireValue(JToken? value, bool truncated, long? byteCount) + { + Value = value; + Truncated = truncated; + ByteCount = byteCount; + } + + public JToken? Value { get; } + public bool Truncated { get; } + public long? ByteCount { get; } + } + + /// + /// Produces the native JSON token for an eval/subscription wire value so consumers + /// receive properly typed output. Values whose serialized form exceeds + /// bytes are reported as truncated with a + /// null value rather than emitting partial, invalid JSON. + /// + public static WireValue ToWireValue(object? value, ReplConfig config) => + ToWireValue(Serialize(value, config), config); + + /// + /// Builds the wire value from an already-serialized JSON string, so callers that + /// also need the string form (e.g. subscription change-detection) serialize once. + /// + public static WireValue ToWireValue(string serialized, ReplConfig config) + { + var byteCount = Encoding.UTF8.GetByteCount(serialized); + if (config.MaxResultLength > 0 && byteCount > config.MaxResultLength) + return new WireValue(null, truncated: true, byteCount); + + try + { + return new WireValue(JToken.Parse(serialized), truncated: false, byteCount: null); + } + catch (JsonReaderException) + { + // Serialize always returns valid JSON, but stay defensive: fall back to a + // string token rather than throwing on the eval hot path. + return new WireValue( + JValue.CreateString(serialized), + truncated: false, + byteCount: null + ); + } + } + public static string Truncate(string serialized, int maxLength) { if (maxLength <= 0) diff --git a/src/HotRepl.Core/Subscriptions/SubscriptionManager.cs b/src/HotRepl.Core/Subscriptions/SubscriptionManager.cs index fb04e5b..c6d8595 100644 --- a/src/HotRepl.Core/Subscriptions/SubscriptionManager.cs +++ b/src/HotRepl.Core/Subscriptions/SubscriptionManager.cs @@ -4,7 +4,6 @@ using HotRepl.Protocol; using HotRepl.Protocol.Serialization; using HotRepl.Serialization; -using Newtonsoft.Json.Linq; namespace HotRepl.Subscriptions; @@ -121,7 +120,6 @@ private bool DeliverValue(SubscriptionState sub, EvalOutcome outcome, Action 0 && sub.DeliveryCount >= sub.Limit; + var wire = + serialized == null ? default : JsonResultSerializer.ToWireValue(serialized, _config); + send( sub.ConnectionId, ProtocolMessageSerializer.Serialize( @@ -141,7 +142,9 @@ private bool DeliverValue(SubscriptionState sub, EvalOutcome outcome, Action /// Contract tests for JsonResultSerializer. /// -/// The serializer produces JSON. Consumers that need a typed value should -/// parse the string with JSON.parse / JToken.Parse; consumers that only -/// need a display string can use it directly. +/// produces a JSON string used for +/// the journal/history display. +/// produces a native for the eval/subscription wire value so +/// consumers receive properly typed output without a second parse, or a +/// truncation signal when the value exceeds the configured size budget. /// public class ResultSerializerTests { @@ -168,6 +170,56 @@ public void Truncate_ExactLimit_ReturnsUnchanged() Assert.Equal(s, JsonResultSerializer.Truncate(s, 50)); } + // ── ToWireValue (typed wire value) ───────────────────────────────────────── + + [Fact] + public void ToWireValue_Int_ProducesNativeNumberToken() + { + var wire = JsonResultSerializer.ToWireValue(2, _defaults); + Assert.False(wire.Truncated); + Assert.NotNull(wire.Value); + Assert.Equal(JTokenType.Integer, wire.Value!.Type); + Assert.Equal(2, wire.Value!.Value()); + } + + [Fact] + public void ToWireValue_String_ProducesNativeStringToken() + { + var wire = JsonResultSerializer.ToWireValue("Ardenfall", _defaults); + Assert.False(wire.Truncated); + Assert.Equal(JTokenType.String, wire.Value!.Type); + Assert.Equal("Ardenfall", wire.Value!.Value()); + } + + [Fact] + public void ToWireValue_Object_ProducesNativeObjectToken() + { + var wire = JsonResultSerializer.ToWireValue(new { X = 1, Y = "hello" }, _defaults); + Assert.False(wire.Truncated); + Assert.Equal(JTokenType.Object, wire.Value!.Type); + Assert.Equal(1, wire.Value!["X"]!.Value()); + Assert.Equal("hello", wire.Value!["Y"]!.Value()); + } + + [Fact] + public void ToWireValue_OversizedValue_TruncatesWithoutValue() + { + var config = new ReplConfig { MaxResultLength = 16 }; + var wire = JsonResultSerializer.ToWireValue(new string('x', 500), config); + Assert.True(wire.Truncated); + Assert.Null(wire.Value); + Assert.NotNull(wire.ByteCount); + Assert.True(wire.ByteCount > 16); + } + + [Fact] + public void ToWireValue_Null_ProducesNullTokenNotTruncated() + { + var wire = JsonResultSerializer.ToWireValue((object?)null, _defaults); + Assert.False(wire.Truncated); + Assert.True(wire.Value == null || wire.Value.Type == JTokenType.Null); + } + // ── Helpers ─────────────────────────────────────────────────────────────── /// Simulates an object whose ToString throws, to exercise the error path. From c498604b0298950d4adf159d4e1718838b5f4722 Mon Sep 17 00:00:00 2001 From: Johann Glock <11704293+glockyco@users.noreply.github.com> Date: Thu, 4 Jun 2026 20:17:18 +0200 Subject: [PATCH 2/2] feat(sdk): handle assembly reloads and cancellation The SDK silently dropped assembly_reload pushes and never sent the cancel frame, so hot reloads left stale command schemas cached and there was no way to stop an eval or an unwanted subscription. Surface reloads via Session.onAssemblyReload (invalidating the cached catalog and descriptors), add Session.cancel(targetId), and have watch() cancel its server subscription automatically when the iterator stops before final. Wire the server side so a cancel targeting a subscription id actually removes the subscription during the tick, not just queued/running evals. FakeRuntime gains matching helpers so the behavior is covered by tests. --- .changeset/assembly-reload-cancel.md | 11 +++++++ docs/control-plane-protocol.md | 8 ++--- packages/protocol/src/messages.ts | 8 ++--- packages/sdk/src/session.ts | 19 +++++++++++ packages/sdk/src/websocket-transport.ts | 33 ++++++++++++++++++- packages/sdk/test/session.test.ts | 26 +++++++++++++++ packages/sdk/test/websocket-transport.test.ts | 9 ++++- packages/testing/src/fake-runtime.ts | 22 +++++++++++++ src/HotRepl.Core/ReplEngine.cs | 9 +++++ 9 files changed, 135 insertions(+), 10 deletions(-) create mode 100644 .changeset/assembly-reload-cancel.md diff --git a/.changeset/assembly-reload-cancel.md b/.changeset/assembly-reload-cancel.md new file mode 100644 index 0000000..5b0228f --- /dev/null +++ b/.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`). diff --git a/docs/control-plane-protocol.md b/docs/control-plane-protocol.md index 37640ac..149a074 100644 --- a/docs/control-plane-protocol.md +++ b/docs/control-plane-protocol.md @@ -211,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 { @@ -226,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" } diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 1a6c297..5d4ca91 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -301,8 +301,8 @@ export const JournalQueryResultMessageSchema = Type.Object( ); export type JournalQueryResultMessage = Static; -/** 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), @@ -400,8 +400,8 @@ export const SubscribeMessageSchema = Type.Object( ); export type SubscribeMessage = Static; -/** 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), diff --git a/packages/sdk/src/session.ts b/packages/sdk/src/session.ts index e4a28ad..cf5683c 100644 --- a/packages/sdk/src/session.ts +++ b/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): AsyncIterable; readArtifact(ref: ArtifactRef): Promise; onSessionEvicted(listener: (event: SessionEvictedMessage) => void): () => void; + onAssemblyReload?(listener: (event: AssemblyReloadMessage) => void): () => void; + cancel?(targetId: string): void; close(): void; } @@ -125,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; @@ -136,6 +140,12 @@ 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 { @@ -143,6 +153,15 @@ export class Session { 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( name: string, args: unknown, diff --git a/packages/sdk/src/websocket-transport.ts b/packages/sdk/src/websocket-transport.ts index efd08b4..97ebb42 100644 --- a/packages/sdk/src/websocket-transport.ts +++ b/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(); private readonly subscriptions = new Map>(); 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(); 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) { diff --git a/packages/sdk/test/session.test.ts b/packages/sdk/test/session.test.ts index 43c2e6f..f151f05 100644 --- a/packages/sdk/test/session.test.ts +++ b/packages/sdk/test/session.test.ts @@ -76,6 +76,32 @@ describe("Session", () => { 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( diff --git a/packages/sdk/test/websocket-transport.test.ts b/packages/sdk/test/websocket-transport.test.ts index 9aabc05..9ab15f5 100644 --- a/packages/sdk/test/websocket-transport.test.ts +++ b/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)); diff --git a/packages/testing/src/fake-runtime.ts b/packages/testing/src/fake-runtime.ts index 2ccac9d..eebe3d6 100644 --- a/packages/testing/src/fake-runtime.ts +++ b/packages/testing/src/fake-runtime.ts @@ -1,5 +1,6 @@ import { type ArtifactRef, + type AssemblyReloadMessage, type CommandDescriptor, defaultLimits, ERROR_KINDS, @@ -74,6 +75,8 @@ export class FakeRuntime implements RuntimeTransport { private readonly journalEntries: JournalEntry[] = []; private readonly counters = new Map(); private readonly evictionListeners = new Set<(event: SessionEvictedMessage) => void>(); + private readonly reloadListeners = new Set<(event: AssemblyReloadMessage) => void>(); + private readonly cancelledTargets: string[] = []; private evalHandler: (code: string) => FakeEvalResult = () => ({}); private activeRequests = 0; private nextJob = 0; @@ -247,6 +250,25 @@ export class FakeRuntime implements RuntimeTransport { for (const listener of this.evictionListeners) listener(event); } + onAssemblyReload(listener: (event: AssemblyReloadMessage) => void): () => void { + this.reloadListeners.add(listener); + return () => this.reloadListeners.delete(listener); + } + + emitAssemblyReload(message = "reloaded", assembly?: string): void { + const event: AssemblyReloadMessage = { type: MESSAGE_TYPES.assemblyReload, message }; + if (assembly !== undefined) event.assembly = assembly; + for (const listener of this.reloadListeners) listener(event); + } + + cancel(targetId: string): void { + this.cancelledTargets.push(targetId); + } + + cancelled(): readonly string[] { + return this.cancelledTargets; + } + async putArtifact( name: string, bytes: Uint8Array, diff --git a/src/HotRepl.Core/ReplEngine.cs b/src/HotRepl.Core/ReplEngine.cs index 0f3d0ed..28d375f 100644 --- a/src/HotRepl.Core/ReplEngine.cs +++ b/src/HotRepl.Core/ReplEngine.cs @@ -163,6 +163,15 @@ public void Tick() // 3. At most one eval per Tick. DrainOneEval(); + // 3b. Cancel subscriptions targeted by a cancel request this Tick. (Eval + // cancels were already applied above; ids that match no subscription + // are ignored.) + if (!_cancelledIds.IsEmpty) + { + foreach (var cancelledId in _cancelledIds.Keys) + _subscriptions!.Cancel(cancelledId); + } + // 4. Subscriptions. _subscriptions!.Tick( (id, code, timeoutMs) => GuardedEvaluate(id, code, timeoutMs),