feat(sim): behaviours, session model, engine

app/src/sim/behavior.ts

@@ -0,0 +1,150 @@
+/// The canned server behaviours a simulated instance can run for one function.
+/// Each is a spec — a label, when it applies, a default config, and a factory
+/// that closes over the config to produce a runnable `Behavior`.
+
+import type { Behavior, SimOutcome } from "./generic.ts";
+import type { FnShape, SchemaShape } from "./shape.ts";
+import { zeroValue } from "./shape.ts";
+
+export type BehaviorKind = "reply" | "echo" | "increment" | "delay" | "raise" | "drop";
+
+export interface BehaviorSpec {
+  kind: BehaviorKind;
+  label: string;
+  /** Whether this behaviour makes sense for `fn` (e.g. `raise` needs throws). */
+  appliesTo(fn: FnShape): boolean;
+  /** A starting config for `fn`, seeded from its return / error types. */
+  defaultConfig(fn: FnShape, schema: SchemaShape): Record<string, unknown>;
+  /** Build the runnable behaviour. */
+  make(config: Record<string, unknown>, fn: FnShape, schema: SchemaShape): Behavior;
+}
+
+const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));
+
+/** `ok` for a normal function, `none` for a one-way one. */
+function okOrNone(fn: FnShape, value: unknown): SimOutcome {
+  return fn.oneway ? { kind: "none" } : { kind: "ok", value };
+}
+
+function getAt(obj: unknown, path: string): unknown {
+  return path.split(".").reduce<unknown>((o, k) => (o == null ? undefined : (o as Record<string, unknown>)[k]), obj);
+}
+function setAt(obj: Record<string, unknown>, path: string, value: unknown): void {
+  const keys = path.split(".");
+  const last = keys.pop()!;
+  let cur = obj;
+  for (const k of keys) cur = (cur[k] ??= {}) as Record<string, unknown>;
+  cur[last] = value;
+}
+/** The first `prim` numeric field path in a struct return, for `increment`. */
+function firstNumericPath(fn: FnShape, schema: SchemaShape): string | undefined {
+  const ret = fn.returns;
+  if (ret?.kind !== "ref") return undefined;
+  const def = schema.types.find((t) => t.name === ret.name);
+  if (def?.kind !== "struct") return undefined;
+  const num = def.fields.find(
+    (f) => f.ty.kind === "prim" && /^[us](8|16|32|64|128)$|^f(32|64)$|^float$/.test(f.ty.name),
+  );
+  return num?.name;
+}
+
+export const BEHAVIORS: Record<BehaviorKind, BehaviorSpec> = {
+  reply: {
+    kind: "reply",
+    label: "Reply with value",
+    appliesTo: () => true,
+    defaultConfig: (fn, schema) => ({
+      value: fn.returns ? zeroValue(fn.returns, schema.types) : null,
+    }),
+    make: (config, fn) => ({
+      run: () => okOrNone(fn, config.value ?? null),
+    }),
+  },
+
+  echo: {
+    kind: "echo",
+    label: "Echo params",
+    appliesTo: () => true,
+    defaultConfig: () => ({}),
+    make: (_config, fn) => ({
+      run: (ctx) => okOrNone(fn, ctx.params),
+    }),
+  },
+
+  increment: {
+    kind: "increment",
+    label: "Increment field",
+    appliesTo: (fn) => !fn.oneway && fn.returns?.kind === "ref",
+    defaultConfig: (fn, schema) => ({
+      base: fn.returns ? zeroValue(fn.returns, schema.types) : {},
+      path: firstNumericPath(fn, schema) ?? "",
+    }),
+    make: (config, fn) => {
+      const path = String(config.path ?? "");
+      let current: Record<string, unknown> | null = null;
+      return {
+        run: () => {
+          current ??= structuredClone(config.base ?? {}) as Record<string, unknown>;
+          if (path) {
+            const n = getAt(current, path);
+            setAt(current, path, (typeof n === "number" ? n : 0) + 1);
+          }
+          return okOrNone(fn, structuredClone(current));
+        },
+      };
+    },
+  },
+
+  delay: {
+    kind: "delay",
+    label: "Delay then reply",
+    appliesTo: () => true,
+    defaultConfig: (fn, schema) => ({
+      ms: 400,
+      value: fn.returns ? zeroValue(fn.returns, schema.types) : null,
+    }),
+    make: (config, fn) => ({
+      run: async () => {
+        await sleep(Number(config.ms) || 0);
+        return okOrNone(fn, config.value ?? null);
+      },
+    }),
+  },
+
+  raise: {
+    kind: "raise",
+    label: "Raise error",
+    appliesTo: (fn) => fn.throws.length > 0,
+    defaultConfig: (fn, schema) => {
+      const first = fn.throws[0];
+      const err = schema.errors.find((e) => e.ordinal === first?.ordinal);
+      const data: Record<string, unknown> = {};
+      for (const f of err?.fields ?? []) data[f.name] = zeroValue(f.ty, schema.types);
+      return { ordinal: first?.ordinal ?? 0, data };
+    },
+    make: (config) => ({
+      run: () => ({
+        kind: "err",
+        ordinal: Number(config.ordinal) || 0,
+        data: config.data ?? null,
+      }),
+    }),
+  },
+
+  drop: {
+    kind: "drop",
+    label: "Drop (never reply)",
+    appliesTo: () => true,
+    defaultConfig: () => ({}),
+    // Never settles — the client's `call` stays pending, like a hung peer.
+    // The promise is released when the connection is closed and discarded.
+    make: () => ({ run: () => new Promise<SimOutcome>(() => {}) }),
+  },
+};
+
+/** The behaviour a freshly-added server function starts on. */
+export function defaultKindFor(fn: FnShape): BehaviorKind {
+  return fn.oneway ? "drop" : "reply";
+}
+
+export const BEHAVIOR_KINDS = Object.keys(BEHAVIORS) as BehaviorKind[];

app/src/sim/engine.test.ts

@@ -0,0 +1,139 @@
+/// Milestone 1c acceptance, scripted end to end: build a session, add a server
+/// and a client, connect, and exercise the behaviours — Reply, Echo, Increment,
+/// Raise error, Drop — plus the model ops (naming, removal, rebuild).
+
+import { readFileSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+import { strict as assert } from "node:assert";
+import { test } from "node:test";
+
+import initWasm, { describe_project } from "../wasm/comline_playground_wasm.js";
+import type { ProjectShape } from "./shape.ts";
+import { addInstance, emptySession, rebuild, removeInstance, setBehavior, setConnection } from "./model.ts";
+import { connect } from "./engine.ts";
+import { SimRemoteError } from "./generic.ts";
+
+await initWasm(
+  readFileSync(fileURLToPath(new URL("../wasm/comline_playground_wasm_bg.wasm", import.meta.url))),
+);
+
+const CHAT = `struct Message {
+    body: string
+    seq: u64
+}
+
+error Rejected {
+    message = "rejected"
+    reason: string
+}
+
+protocol Chat {
+    function send(text: string) -> Message ! Rejected;
+    function note(text: string);
+}
+`;
+
+const shape = () => describe_project([{ path: "chat.ids", source: CHAT }]) as ProjectShape;
+
+function wired() {
+  const session = emptySession(shape());
+  const server = addInstance(session, { schemaNs: "chat", protocol: "Chat", role: "server" });
+  const client = addInstance(session, { schemaNs: "chat", protocol: "Chat", role: "client" });
+  setConnection(session, client.id, server.id);
+  return { session, server, client };
+}
+
+test("model — instance naming, seeded behaviours, removal drops the connection", () => {
+  const session = emptySession(shape());
+  const a = addInstance(session, { schemaNs: "chat", protocol: "Chat", role: "server" });
+  const b = addInstance(session, { schemaNs: "chat", protocol: "Chat", role: "server" });
+  assert.deepEqual([a.name, b.name], ["chat-1", "chat-2"]);
+  assert.deepEqual(Object.keys(a.behaviors).sort(), ["note", "send"]);
+  assert.equal(a.behaviors.send.kind, "reply");
+  assert.equal(a.behaviors.note.kind, "drop"); // one-way default
+
+  const client = addInstance(session, { schemaNs: "chat", protocol: "Chat", role: "client" });
+  assert.deepEqual(client.behaviors, {});
+  setConnection(session, client.id, a.id);
+  removeInstance(session, a.id);
+  assert.equal(session.connection, null);
+});
+
+test("Reply with value returns the configured value", async () => {
+  const { session, server } = wired();
+  setBehavior(session, server.id, "send", "reply", { value: { body: "HI", seq: 1 } });
+  const live = await connect(session, session.connection!);
+
+  assert.deepEqual(await live.call("send", { text: "x" }), { body: "HI", seq: 1 });
+  assert.equal(live.tap.frames.filter((f) => f.kind !== "handshake").length, 2);
+  live.close();
+});
+
+test("Echo returns the params; a live behaviour swap takes effect on the next call", async () => {
+  const { session } = wired();
+  const live = await connect(session, session.connection!);
+
+  live.setBehavior("send", { kind: "echo", config: {} });
+  assert.deepEqual(await live.call("send", { text: "pong" }), { text: "pong" });
+
+  live.setBehavior("send", { kind: "reply", config: { value: { body: "z", seq: 9 } } });
+  assert.deepEqual(await live.call("send", { text: "x" }), { body: "z", seq: 9 });
+  live.close();
+});
+
+test("Increment field bumps once per call", async () => {
+  const { session, server } = wired();
+  setBehavior(session, server.id, "send", "increment", {
+    base: { body: "b", seq: 0 },
+    path: "seq",
+  });
+  const live = await connect(session, session.connection!);
+
+  assert.equal((await live.call("send", { text: "x" }) as { seq: number }).seq, 1);
+  assert.equal((await live.call("send", { text: "x" }) as { seq: number }).seq, 2);
+  assert.equal((await live.call("send", { text: "x" }) as { seq: number }).seq, 3);
+  live.close();
+});
+
+test("Raise error comes back as SimRemoteError mapped to the ordinal's name", async () => {
+  const { session, server } = wired();
+  setBehavior(session, server.id, "send", "raise", { ordinal: 0, data: { reason: "denied" } });
+  const live = await connect(session, session.connection!);
+
+  await assert.rejects(
+    () => live.call("send", { text: "x" }),
+    (e: unknown) => {
+      assert.ok(e instanceof SimRemoteError);
+      assert.equal(e.ordinal, 0);
+      assert.equal(e.errorName, "Rejected");
+      assert.deepEqual(e.data, { reason: "denied" });
+      return true;
+    },
+  );
+  live.close();
+});
+
+test("Drop leaves the call pending", async () => {
+  const { session } = wired();
+  const live = await connect(session, session.connection!);
+  live.setBehavior("send", { kind: "drop", config: {} });
+
+  const race = await Promise.race([
+    live.call("send", { text: "x" }).then(() => "settled"),
+    new Promise((r) => setTimeout(() => r("pending"), 60)),
+  ]);
+  assert.equal(race, "pending");
+  assert.equal(live.tap.frames.filter((f) => f.kind !== "handshake" && f.from === "server").length, 0);
+  live.close();
+});
+
+test("rebuild keeps a surviving instance's behaviour config", () => {
+  const { session, server } = wired();
+  setBehavior(session, server.id, "send", "raise", { ordinal: 0, data: { reason: "keep me" } });
+  rebuild(session, shape()); // same schema, recompiled
+
+  const kept = session.instances.find((i) => i.id === server.id)!;
+  assert.equal(kept.behaviors.send.kind, "raise");
+  assert.deepEqual(kept.behaviors.send.config, { ordinal: 0, data: { reason: "keep me" } });
+  assert.ok(session.connection, "connection survives a same-shape rebuild");
+});

app/src/sim/engine.ts

@@ -0,0 +1,75 @@
+/// Turn a `Connection` into a running wire: a `GenericDispatch` server bound to
+/// the client instance's behaviours, a `GenericClient` on the other end, both
+/// over a tapped `duplex()`. Behaviours can be swapped without reconnecting.
+
+import { BEHAVIORS } from "./behavior.ts";
+import { GenericClient, GenericDispatch, type Behavior, type BehaviorMap } from "./generic.ts";
+import { instance, type Connection, type Session } from "./model.ts";
+import {
+  DatagramFraming,
+  Handshake,
+  JsonCodec,
+  JsonRpcFraming,
+  Server,
+  Client,
+  type Framing,
+} from "./runtime/index.ts";
+import { findProtocol } from "./shape.ts";
+import { wire, type Tap } from "./transport.ts";
+
+export interface LiveConnection {
+  tap: Tap;
+  /** Invoke a function from the client side. */
+  call(fnName: string, params: unknown): Promise<unknown>;
+  /** Swap a server behaviour; takes effect on the next call. */
+  setBehavior(fnName: string, setting: { kind: keyof typeof BEHAVIORS; config: Record<string, unknown> }): void;
+  /** Drop both ends and end the serve loop. */
+  close(): void;
+}
+
+const framingFor = (name: "datagram" | "jsonrpc"): (() => Framing) =>
+  name === "jsonrpc" ? () => new JsonRpcFraming() : () => new DatagramFraming();
+
+/** Bind a `Connection` to real transports and run its handshake. */
+export async function connect(session: Session, conn: Connection): Promise<LiveConnection> {
+  const client = instance(session, conn.clientId);
+  const server = instance(session, conn.serverId);
+  if (!client || !server) throw new Error("connect: unknown instance");
+
+  const found = findProtocol(session.shape, server.schemaNs, server.protocol);
+  if (!found) throw new Error(`connect: ${server.schemaNs}::${server.protocol} is not compiled`);
+  const { schema, protocol } = found;
+
+  const codec = new JsonCodec();
+  const makeFraming = framingFor(protocol.framing);
+  const handshake = () =>
+    new Handshake({ irHash: BigInt(schema.ir_hash), wireFormat: codec.name, framing: makeFraming().name });
+
+  // A mutable map so `setBehavior` can swap an entry live.
+  const behaviors: BehaviorMap = {};
+  const build = (fnName: string): Behavior => {
+    const fn = protocol.functions.find((f) => f.name === fnName)!;
+    const setting = server.behaviors[fnName] ?? { kind: "reply" as const, config: {} };
+    return BEHAVIORS[setting.kind].make(setting.config, fn, schema);
+  };
+  for (const fn of protocol.functions) behaviors[fn.name] = build(fn.name);
+
+  const w = wire(client.name, server.name);
+  const rpcServer = new Server(new GenericDispatch(protocol, behaviors), codec, makeFraming());
+  void rpcServer.serveHandshaked(w.b, handshake());
+  const rpcClient = new GenericClient(
+    await Client.connect(w.a, codec, handshake(), makeFraming()),
+    protocol,
+  );
+
+  return {
+    tap: w.tap,
+    call: (fnName, params) => rpcClient.call(fnName, params),
+    setBehavior: (fnName, setting) => {
+      const fn = protocol.functions.find((f) => f.name === fnName);
+      if (!fn) throw new Error(`setBehavior: no function ${fnName}`);
+      behaviors[fnName] = BEHAVIORS[setting.kind].make(setting.config, fn, schema);
+    },
+    close: () => w.close(),
+  };
+}