refactor: Replace VatWorkerManager message types with new RPC pattern

packages/extension/package.json

@@ -43,8 +43,8 @@
   "dependencies": {
     "@endo/eventual-send": "^1.3.1",
     "@endo/marshal": "^1.6.4",
-    "@endo/promise-kit": "^1.1.10",
     "@metamask/json-rpc-engine": "^10.0.3",
+    "@metamask/rpc-errors": "^7.0.2",
     "@metamask/snaps-utils": "^9.1.0",
     "@metamask/superstruct": "^3.2.0",
     "@metamask/utils": "^11.4.0",

packages/extension/src/kernel-integration/VatWorkerClient.test.ts

@@ -1,9 +1,10 @@
-import type { VatId, VatWorkerServiceReply, VatConfig } from '@ocap/kernel';
-import { VatWorkerServiceCommandMethod } from '@ocap/kernel';
+import { rpcErrors } from '@metamask/rpc-errors';
+import type { JsonRpcResponse } from '@metamask/utils';
+import type { VatId, VatConfig } from '@ocap/kernel';
 import type { PostMessageTarget } from '@ocap/streams/browser';
 import { TestDuplexStream } from '@ocap/test-utils/streams';
 import type { Logger } from '@ocap/utils';
-import { delay, makeLogger } from '@ocap/utils';
+import { delay, makeLogger, stringify } from '@ocap/utils';
 import { describe, it, expect, beforeEach, vi } from 'vitest';
 
 import type { VatWorkerClientStream } from './VatWorkerClient.ts';
@@ -38,39 +39,28 @@ const makeVatConfig = (sourceSpec: string = 'bogus.js'): VatConfig => ({
   sourceSpec,
 });
 
-const makeMessageEvent = (
+const makeMessageEvent = <Response extends Partial<JsonRpcResponse>>(
   messageId: `m${number}`,
-  payload: VatWorkerServiceReply['payload'],
+  payload: Response,
   port?: MessagePort,
-): MessageEvent =>
+): MessageEvent<Response> =>
   new MessageEvent('message', {
-    data: { id: messageId, payload },
+    data: { ...payload, id: messageId, jsonrpc: '2.0' },
     ports: port ? [port] : [],
   });
 
-const makeLaunchReply = (messageId: `m${number}`, vatId: VatId): MessageEvent =>
+const makeLaunchReply = (messageId: `m${number}`): MessageEvent =>
   makeMessageEvent(
     messageId,
     {
-      method: VatWorkerServiceCommandMethod.launch,
-      params: { vatId },
+      result: null,
     },
     new MessageChannel().port1,
   );
 
-const makeTerminateReply = (
-  messageId: `m${number}`,
-  vatId: VatId,
-): MessageEvent =>
-  makeMessageEvent(messageId, {
-    method: VatWorkerServiceCommandMethod.terminate,
-    params: { vatId },
-  });
-
-const makeTerminateAllReply = (messageId: `m${number}`): MessageEvent =>
+const makeNullReply = (messageId: `m${number}`): MessageEvent =>
   makeMessageEvent(messageId, {
-    method: VatWorkerServiceCommandMethod.terminateAll,
-    params: null,
+    result: null,
   });
 
 describe('ExtensionVatWorkerClient', () => {
@@ -107,58 +97,30 @@ describe('ExtensionVatWorkerClient', () => {
       });
     });
 
-    it('logs error for unexpected methods', async () => {
-      const errorSpy = vi.spyOn(clientLogger, 'error');
-      client.launch('v0', makeVatConfig()).catch((error) => {
-        throw error;
-      });
-      // @ts-expect-error Destructive testing.
-      await stream.receiveInput(makeMessageEvent('m1', { method: 'foo' }));
-      await delay(10);
-
-      expect(errorSpy).toHaveBeenCalled();
-      expect(errorSpy).toHaveBeenCalledWith(
-        'Received message with unexpected method',
-        'foo',
-      );
-    });
-
     it('rejects pending promises for error replies', async () => {
       const resultP = client.launch('v0', makeVatConfig());
 
       await stream.receiveInput(
         makeMessageEvent('m1', {
-          method: VatWorkerServiceCommandMethod.launch,
-          params: { vatId: 'v0', error: new Error('foo') },
+          error: rpcErrors.internal('foo'),
         }),
       );
 
       await expect(resultP).rejects.toThrow('foo');
     });
 
-    it.each`
-      method
-      ${VatWorkerServiceCommandMethod.launch}
-      ${VatWorkerServiceCommandMethod.terminate}
-    `(
-      "calls logger.error when receiving a $method reply it wasn't waiting for",
-      async ({ method }) => {
-        const errorSpy = vi.spyOn(clientLogger, 'error');
-        const unexpectedReply = makeMessageEvent('m9', {
-          method,
-          params: { vatId: 'v0' },
-        });
-
-        await stream.receiveInput(unexpectedReply);
-        await delay(10);
+    it('calls logger.error when receiving an unexpected reply', async () => {
+      const errorSpy = vi.spyOn(clientLogger, 'error');
+      const unexpectedReply = makeNullReply('m9');
 
-        expect(errorSpy).toHaveBeenCalledOnce();
-        expect(errorSpy).toHaveBeenLastCalledWith(
-          'Received unexpected reply',
-          unexpectedReply.data,
-        );
-      },
-    );
+      await stream.receiveInput(unexpectedReply);
+      await delay(10);
+
+      expect(errorSpy).toHaveBeenCalledOnce();
+      expect(errorSpy).toHaveBeenLastCalledWith(
+        'Received response with unexpected id "m9".',
+      );
+    });
 
     describe('launch', () => {
       it('resolves with a duplex stream when receiving a launch reply', async () => {
@@ -167,39 +129,29 @@ describe('ExtensionVatWorkerClient', () => {
         const result = client.launch(vatId, vatConfig);
 
         await delay(10);
-        await stream.receiveInput(makeLaunchReply('m1', vatId));
+        await stream.receiveInput(makeLaunchReply('m1'));
 
         // @ocap/streams is mocked
         expect(await result).toBeInstanceOf(TestDuplexStream);
       });
 
-      it('logs error when receiving reply without a port', async () => {
-        const errorSpy = vi.spyOn(clientLogger, 'error');
+      it('throws an error when receiving reply without a port', async () => {
         const vatId: VatId = 'v0';
         const vatConfig = makeVatConfig();
-        client.launch(vatId, vatConfig).catch((error) => {
-          throw error;
-        });
-        const reply = makeMessageEvent('m1', {
-          method: VatWorkerServiceCommandMethod.launch,
-          params: { vatId },
-        });
+        const launchP = client.launch(vatId, vatConfig);
+        const reply = makeNullReply('m1');
 
         await stream.receiveInput(reply);
-        await delay(10);
-
-        expect(errorSpy).toHaveBeenCalledOnce();
-        expect(errorSpy.mock.lastCall?.[0]).toBe(
-          'Expected a port with message reply',
+        await expect(launchP).rejects.toThrow(
+          `No port found for launch of: ${stringify({ vatId, vatConfig })}`,
         );
-        expect(errorSpy.mock.lastCall?.[1]).toBe(reply);
       });
     });
 
     describe('terminate', () => {
       it('resolves when receiving a terminate reply', async () => {
         const result = client.terminate('v0');
-        await stream.receiveInput(makeTerminateReply('m1', 'v0'));
+        await stream.receiveInput(makeNullReply('m1'));
         await delay(10);
 
         expect(await result).toBeUndefined();
@@ -209,7 +161,7 @@ describe('ExtensionVatWorkerClient', () => {
     describe('terminateAll', () => {
       it('resolves when receiving a terminateAll reply', async () => {
         const result = client.terminateAll();
-        await stream.receiveInput(makeTerminateAllReply('m1'));
+        await stream.receiveInput(makeNullReply('m1'));
         await delay(10);
 
         expect(await result).toBeUndefined();