feat(repositories): refuse partial provider connectors and cover RPC dispatch

Reject incomplete worker connectors at registration time and verify implemented and missing hooks through the real JSON-RPC worker host.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/plugins/sdk/tests/repository-provider-rpc.test.ts

@@ -0,0 +1,146 @@
+import { createInterface } from "node:readline";
+import { PassThrough } from "node:stream";
+import { describe, expect, it } from "vitest";
+import { definePlugin } from "../src/define-plugin.js";
+import {
+  createRequest,
+  isJsonRpcResponse,
+  parseMessage,
+  serializeMessage,
+  type JsonRpcResponse,
+} from "../src/protocol.js";
+import { startWorkerRpcHost } from "../src/worker-rpc-host.js";
+
+/**
+ * Exercises the repository-provider hooks through the real JSON-RPC worker
+ * host, not just as function calls: a hook that is defined must be advertised
+ * in `initialize` and reachable by its method name, and one that is not defined
+ * must fail closed rather than resolve to undefined.
+ */
+
+const HOST = "github.acme-corp.example";
+const PROVIDER = "github-enterprise";
+const COMPANY_ID = "11111111-1111-4111-8111-111111111111";
+
+function makeWorker(plugin: ReturnType<typeof definePlugin>) {
+  const hostToWorker = new PassThrough();
+  const workerToHost = new PassThrough();
+  const hostReadline = createInterface({ input: workerToHost });
+  const pending = new Map<string, (response: JsonRpcResponse) => void>();
+  let nextRequestId = 1;
+
+  const worker = startWorkerRpcHost({ plugin, stdin: hostToWorker, stdout: workerToHost });
+
+  hostReadline.on("line", (line) => {
+    const message = parseMessage(line);
+    if (!isJsonRpcResponse(message)) return;
+    pending.get(String(message.id))?.(message);
+    pending.delete(String(message.id));
+  });
+
+  function callWorker(method: string, params: unknown) {
+    const id = `host-${nextRequestId++}`;
+    const result = new Promise<unknown>((resolve, reject) => {
+      pending.set(id, (response) => {
+        if ("error" in response && response.error) {
+          reject(Object.assign(new Error(response.error.message), { code: response.error.code }));
+          return;
+        }
+        resolve((response as { result?: unknown }).result);
+      });
+    });
+    hostToWorker.write(serializeMessage(createRequest(method, params, id)));
+    return result;
+  }
+
+  async function initialize() {
+    return await callWorker("initialize", {
+      manifest: {
+        id: "acme.github-enterprise",
+        apiVersion: 1,
+        version: "1.0.0",
+        displayName: "GitHub Enterprise",
+        description: "Test provider",
+        author: "Paperclip",
+        categories: ["connector"],
+        capabilities: ["repository.providers.register"],
+        entrypoints: { worker: "dist/worker.js" },
+        repositoryProviders: [{ providerKey: PROVIDER, displayName: "GitHub Enterprise", host: HOST }],
+      },
+      config: {},
+      databaseNamespace: null,
+    });
+  }
+
+  function stop() {
+    worker.stop();
+    hostReadline.close();
+    hostToWorker.destroy();
+    workerToHost.destroy();
+  }
+
+  return { callWorker, initialize, stop };
+}
+
+describe("repository provider worker RPC", () => {
+  it("advertises implemented hooks and dispatches to them", async () => {
+    const seen: string[] = [];
+    const plugin = definePlugin({
+      async setup() {},
+      async onRepositoryProviderBeginInstallation(params) {
+        seen.push(`begin:${params.companyId}:${params.host}`);
+        return {
+          installUrl: `https://${HOST}/install?state=abc`,
+          state: "abc",
+          expiresAt: new Date(Date.now() + 60_000).toISOString(),
+        };
+      },
+      async onRepositoryProviderDiscover(params) {
+        seen.push(`discover:${params.connection.id}`);
+        return { items: [], nextCursor: null, total: 0 };
+      },
+    });
+    const { callWorker, initialize, stop } = makeWorker(plugin);
+
+    try {
+      const initialized = await initialize() as { supportedMethods: string[] };
+      expect(initialized.supportedMethods).toContain("repositoryProviderBeginInstallation");
+      expect(initialized.supportedMethods).toContain("repositoryProviderDiscover");
+      // Not implemented by this plugin, so not advertised — the host uses this
+      // to refuse registration of a partially implemented connector.
+      expect(initialized.supportedMethods).not.toContain("repositoryProviderResolveCloneCredential");
+
+      const begun = await callWorker("repositoryProviderBeginInstallation", {
+        providerKey: PROVIDER,
+        host: HOST,
+        companyId: COMPANY_ID,
+        userId: "22222222-2222-4222-8222-222222222222",
+        redirectPath: null,
+      }) as { state: string };
+      expect(begun.state).toBe("abc");
+      expect(seen).toContain(`begin:${COMPANY_ID}:${HOST}`);
+    } finally {
+      stop();
+    }
+  });
+
+  it("fails closed when a repository provider method is not implemented", async () => {
+    const plugin = definePlugin({ async setup() {} });
+    const { callWorker, initialize, stop } = makeWorker(plugin);
+
+    try {
+      await initialize();
+      await expect(
+        callWorker("repositoryProviderResolveCloneCredential", {
+          providerKey: PROVIDER,
+          host: HOST,
+          companyId: COMPANY_ID,
+          connection: {},
+          repository: {},
+        }),
+      ).rejects.toThrow(/repositoryProviderResolveCloneCredential/);
+    } finally {
+      stop();
+    }
+  });
+});

server/src/__tests__/repository-provider-plugin-seam.test.ts

@@ -13,6 +13,7 @@ import {
   listAvailableRepositoryProviders,
   markCloneCredentialNoStore,
   pluginRepositoryProviderOwnerKey,
+  REQUIRED_REPOSITORY_PROVIDER_METHODS,
   type RepositoryProviderPluginRecord,
   type RepositoryProviderPluginSource,
 } from "../services/repository-providers/index.js";
@@ -81,6 +82,8 @@ function fakeLifecycle() {
 
 interface FakeWorkerOptions {
   running?: boolean;
+  /** Methods the worker reported at initialization. Omit to skip the check. */
+  supportedMethods?: string[];
   handlers?: Partial<{
     [M in HostToWorkerMethodName]: (params: HostToWorkerMethods[M][0]) => unknown;
   }>;
@@ -95,6 +98,10 @@ function fakeWorkerManager(options: FakeWorkerOptions = {}) {
       running = next;
     },
     isRunning: () => running,
+    getWorker: ((() =>
+      (options.supportedMethods
+        ? { supportedMethods: options.supportedMethods }
+        : undefined)) as never),
     call: (async (_pluginId: string, method: HostToWorkerMethodName, params: unknown) => {
       calls.push({ method, params });
       const handler = options.handlers?.[method] as ((p: unknown) => unknown) | undefined;
@@ -219,6 +226,36 @@ describe("registration lifecycle", () => {
     registrar.stop();
   });
 
+  it("does not register a worker that implements only part of the connector", async () => {
+    const lifecycle = fakeLifecycle();
+    const registrar = createPluginRepositoryProviderRegistrar({
+      lifecycle,
+      workerManager: fakeWorkerManager({
+        supportedMethods: [...REQUIRED_REPOSITORY_PROVIDER_METHODS].filter(
+          (method) => method !== "repositoryProviderResolveCloneCredential",
+        ),
+      }),
+      plugins: pluginSource([pluginRecord()]),
+    });
+    await registrar.start();
+
+    expect(getRepositoryProviderConnector(PROVIDER, { host: HOST })).toBeNull();
+    registrar.stop();
+  });
+
+  it("registers a worker that implements the whole connector", async () => {
+    const lifecycle = fakeLifecycle();
+    const registrar = createPluginRepositoryProviderRegistrar({
+      lifecycle,
+      workerManager: fakeWorkerManager({ supportedMethods: [...REQUIRED_REPOSITORY_PROVIDER_METHODS] }),
+      plugins: pluginSource([pluginRecord()]),
+    });
+    await registrar.start();
+
+    expect(getRepositoryProviderConnector(PROVIDER, { host: HOST })).not.toBeNull();
+    registrar.stop();
+  });
+
   it("does not register a plugin that is installed but not ready", async () => {
     const lifecycle = fakeLifecycle();
     const registrar = createPluginRepositoryProviderRegistrar({

server/src/services/repository-providers/plugin-connector.ts

@@ -38,7 +38,38 @@ import type {
  */
 
 /** Worker-manager surface the bridge needs; narrow for easy testing. */
-export type RepositoryProviderWorkerHost = Pick<PluginWorkerManager, "call" | "isRunning">;
+export type RepositoryProviderWorkerHost = Pick<PluginWorkerManager, "call" | "isRunning"> & {
+  getWorker?: PluginWorkerManager["getWorker"];
+};
+
+/**
+ * The hooks a worker must implement before the host will treat it as a
+ * repository provider. A connector that answers installation but not credential
+ * resolution would strand an operator halfway through connecting a repository,
+ * so the seam refuses the registration instead.
+ */
+export const REQUIRED_REPOSITORY_PROVIDER_METHODS = [
+  "repositoryProviderBeginInstallation",
+  "repositoryProviderCompleteInstallation",
+  "repositoryProviderDiscover",
+  "repositoryProviderRefreshMetadata",
+  "repositoryProviderSync",
+  "repositoryProviderResolveCloneCredential",
+] as const;
+
+/**
+ * Methods a running worker declared at initialization but does not implement.
+ * Returns an empty list when the worker manager cannot report them, since an
+ * older worker handle predates the advertisement and is checked at call time.
+ */
+export function missingRepositoryProviderMethods(
+  workerManager: RepositoryProviderWorkerHost,
+  pluginId: string,
+): string[] {
+  const supported = workerManager.getWorker?.(pluginId)?.supportedMethods;
+  if (!supported) return [];
+  return REQUIRED_REPOSITORY_PROVIDER_METHODS.filter((method) => !supported.includes(method));
+}
 
 export interface PluginRepositoryProviderConnectorOptions {
   pluginId: string;

server/src/services/repository-providers/plugin-registrar.ts

@@ -8,7 +8,11 @@ import { logger } from "../../middleware/logger.js";
 import type { PluginLifecycleManager } from "../plugin-lifecycle.js";
 import { pluginRegistryService } from "../plugin-registry.js";
 import { repositoryProviderRegistry } from "../repository-connections.js";
-import { createPluginRepositoryProviderConnector, type RepositoryProviderWorkerHost } from "./plugin-connector.js";
+import {
+  createPluginRepositoryProviderConnector,
+  missingRepositoryProviderMethods,
+  type RepositoryProviderWorkerHost,
+} from "./plugin-connector.js";
 import { listAvailableRepositoryProviders, type RepositoryProviderAvailability } from "./registry.js";
 
 /**
@@ -137,6 +141,15 @@ export function createPluginRepositoryProviderRegistrar(
     const declarations = authorizedRepositoryProviderDeclarations(plugin.manifestJson);
     if (declarations.length === 0) return 0;
 
+    const missingMethods = missingRepositoryProviderMethods(options.workerManager, pluginId);
+    if (missingMethods.length > 0) {
+      log.warn(
+        { pluginId, pluginKey: plugin.pluginKey, missingMethods },
+        "repository provider registration refused: worker does not implement the full connector",
+      );
+      return 0;
+    }
+
     const ownerKey = pluginRepositoryProviderOwnerKey(pluginId);
     const identities: string[] = [];
     for (const declaration of declarations) {