PaperClipAI/packages/plugins/sdk/src/worker-rpc-host.ts

/**
 * Worker-side RPC host — runs inside the child process spawned by the host.
 *
 * This module is the worker-side counterpart to the server's
 * `PluginWorkerManager`. It:
 *
 * 1. Reads newline-delimited JSON-RPC 2.0 requests from **stdin**
 * 2. Dispatches them to the appropriate plugin handler (events, jobs, tools, …)
 * 3. Writes JSON-RPC 2.0 responses back on **stdout**
 * 4. Provides a concrete `PluginContext` whose SDK client methods (e.g.
 *    `ctx.state.get()`, `ctx.events.emit()`) send JSON-RPC requests to the
 *    host on stdout and await responses on stdin.
 *
 * ## Message flow
 *
 * ```
 * Host (parent)                          Worker (this module)
 *   |                                        |
 *   |--- request(initialize) ------------->  |  → calls plugin.setup(ctx)
 *   |<-- response(ok:true) ----------------  |
 *   |                                        |
 *   |--- notification(onEvent) ----------->  |  → dispatches to registered handler
 *   |                                        |
 *   |<-- request(state.get) ---------------  |  ← SDK client call from plugin code
 *   |--- response(result) ---------------->  |
 *   |                                        |
 *   |--- request(shutdown) --------------->  |  → calls plugin.onShutdown()
 *   |<-- response(void) ------------------  |
 *   |                                        (process exits)
 * ```
 *
 * @see PLUGIN_SPEC.md §12 — Process Model
 * @see PLUGIN_SPEC.md §13 — Host-Worker Protocol
 * @see PLUGIN_SPEC.md §14 — SDK Surface
 */

import fs from "node:fs";
import { AsyncLocalStorage } from "node:async_hooks";
import path from "node:path";
import { createInterface, type Interface as ReadlineInterface } from "node:readline";
import { fileURLToPath } from "node:url";

import type {
  AskUserQuestionsInteraction,
  PaperclipPluginManifestV1,
  RequestCheckboxConfirmationInteraction,
  RequestConfirmationInteraction,
  SuggestTasksInteraction,
} from "@paperclipai/shared";

import type { PaperclipPlugin } from "./define-plugin.js";
import type {
  PluginApiRequestInput,
  PluginHealthDiagnostics,
  PluginConfigValidationResult,
  PluginWebhookInput,
} from "./define-plugin.js";
import type {
  PluginContext,
  PluginEvent,
  PluginJobContext,
  PluginLauncherRegistration,
  ScopeKey,
  ToolRunContext,
  ToolResult,
  EventFilter,
  AgentSessionEvent,
} from "./types.js";
import type {
  JsonRpcId,
  JsonRpcNotification,
  JsonRpcRequest,
  JsonRpcResponse,
  InitializeParams,
  InitializeResult,
  ConfigChangedParams,
  ValidateConfigParams,
  OnEventParams,
  RunJobParams,
  GetDataParams,
  PerformActionParams,
  PluginPerformActionActorContext,
  PluginPerformActionContext,
  ExecuteToolParams,
  DetectExternalObjectsParams,
  ResolveExternalObjectParams,
  RefreshExternalObjectsParams,
  PluginEnvironmentAcquireLeaseParams,
  PluginEnvironmentDestroyLeaseParams,
  PluginEnvironmentExecuteParams,
  PluginEnvironmentRealizeWorkspaceParams,
  PluginEnvironmentReleaseLeaseParams,
  PluginEnvironmentResumeLeaseParams,
  PluginEnvironmentValidateConfigParams,
  PluginEnvironmentProbeParams,
  PluginEnvironmentStartInteractiveSetupParams,
  PluginEnvironmentGetInteractiveSetupParams,
  PluginEnvironmentCaptureTemplateParams,
  PluginEnvironmentCancelInteractiveSetupParams,
  PluginEnvironmentDeleteTemplateParams,
  PluginInvocationContext,
  WorkerToHostMethodName,
  WorkerToHostMethods,
} from "./protocol.js";
import {
  JSONRPC_VERSION,
  JSONRPC_ERROR_CODES,
  PLUGIN_RPC_ERROR_CODES,
  createRequest,
  createSuccessResponse,
  createErrorResponse,
  createNotification,
  parseMessage,
  serializeMessage,
  isJsonRpcRequest,
  isJsonRpcResponse,
  isJsonRpcNotification,
  isJsonRpcSuccessResponse,
  isJsonRpcErrorResponse,
  JsonRpcParseError,
  JsonRpcCallError,
} from "./protocol.js";

// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------

/**
 * Options for starting the worker-side RPC host.
 */
export interface WorkerRpcHostOptions {
  /**
   * The plugin definition returned by `definePlugin()`.
   *
   * The worker entrypoint should import its plugin and pass it here.
   */
  plugin: PaperclipPlugin;

  /**
   * Input stream to read JSON-RPC messages from.
   * Defaults to `process.stdin`.
   */
  stdin?: NodeJS.ReadableStream;

  /**
   * Output stream to write JSON-RPC messages to.
   * Defaults to `process.stdout`.
   */
  stdout?: NodeJS.WritableStream;

  /**
   * Default timeout (ms) for worker→host RPC calls.
   * Defaults to 30 000 ms.
   */
  rpcTimeoutMs?: number;
}

/**
 * A running worker RPC host instance.
 *
 * Returned by `startWorkerRpcHost()`. Callers (usually just the worker
 * bootstrap) hold a reference so they can inspect status or force-stop.
 */
export interface WorkerRpcHost {
  /** Whether the host is currently running and listening for messages. */
  readonly running: boolean;

  /**
   * Stop the RPC host immediately. Closes readline, rejects pending
   * outbound calls, and does NOT call the plugin's shutdown hook (that
   * should have already been called via the `shutdown` RPC method).
   */
  stop(): void;
}

// ---------------------------------------------------------------------------
// Internal: event registration
// ---------------------------------------------------------------------------

interface EventRegistration {
  name: string;
  filter?: EventFilter;
  fn: (event: PluginEvent) => Promise<void>;
}

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

/** Default timeout for worker→host RPC calls. */
const DEFAULT_RPC_TIMEOUT_MS = 30_000;

function realpathOrResolvedPath(filePath: string): string {
  const resolvedPath = path.resolve(filePath);
  try {
    return fs.realpathSync.native(resolvedPath);
  } catch {
    return resolvedPath;
  }
}

export function isWorkerEntrypoint(entry: string, moduleUrl: string): boolean {
  const thisFile = realpathOrResolvedPath(fileURLToPath(moduleUrl));
  const entryPath = realpathOrResolvedPath(entry);
  return thisFile === entryPath;
}

// ---------------------------------------------------------------------------
// startWorkerRpcHost
// ---------------------------------------------------------------------------

/**
 * Options for runWorker when testing (optional stdio to avoid using process streams).
 * When both stdin and stdout are provided, the "is main module" check is skipped
 * and the host is started with these streams. Used by tests.
 */
export interface RunWorkerOptions {
  stdin?: NodeJS.ReadableStream;
  stdout?: NodeJS.WritableStream;
}

/**
 * Start the worker when this module is the process entrypoint.
 *
 * Call this at the bottom of your worker file so that when the host runs
 * `node dist/worker.js`, the RPC host starts and the process stays alive.
 * When the module is imported (e.g. for re-exports or tests), nothing runs.
 *
 * When `options.stdin` and `options.stdout` are provided (e.g. in tests),
 * the main-module check is skipped and the host is started with those streams.
 *
 * @example
 * ```ts
 * const plugin = definePlugin({ ... });
 * export default plugin;
 * runWorker(plugin, import.meta.url);
 * ```
 */
export function runWorker(
  plugin: PaperclipPlugin,
  moduleUrl: string,
  options?: RunWorkerOptions,
): WorkerRpcHost | void {
  if (
    options?.stdin != null &&
    options?.stdout != null
  ) {
    return startWorkerRpcHost({
      plugin,
      stdin: options.stdin,
      stdout: options.stdout,
    });
  }
  const entry = process.argv[1];
  if (typeof entry !== "string") return;
  if (isWorkerEntrypoint(entry, moduleUrl)) {
    startWorkerRpcHost({ plugin });
  }
}

/**
 * Start the worker-side RPC host.
 *
 * This function is typically called from a thin bootstrap script that is the
 * actual entrypoint of the child process:
 *
 * ```ts
 * // worker-bootstrap.ts
 * import plugin from "./worker.js";
 * import { startWorkerRpcHost } from "@paperclipai/plugin-sdk";
 *
 * startWorkerRpcHost({ plugin });
 * ```
 *
 * The host begins listening on stdin immediately. It does NOT call
 * `plugin.definition.setup()` yet — that happens when the host sends the
 * `initialize` RPC.
 *
 * @returns A handle for inspecting or stopping the RPC host
 */
export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost {
  const { plugin } = options;
  const stdinStream = options.stdin ?? process.stdin;
  const stdoutStream = options.stdout ?? process.stdout;
  const rpcTimeoutMs = options.rpcTimeoutMs ?? DEFAULT_RPC_TIMEOUT_MS;

  // -----------------------------------------------------------------------
  // State
  // -----------------------------------------------------------------------

  let running = true;
  let initialized = false;
  let manifest: PaperclipPluginManifestV1 | null = null;
  let currentConfig: Record<string, unknown> = {};
  let databaseNamespace: string | null = null;
  const invocationContextStorage = new AsyncLocalStorage<PluginInvocationContext>();

  // Plugin handler registrations (populated during setup())
  const eventHandlers: EventRegistration[] = [];
  const jobHandlers = new Map<string, (job: PluginJobContext) => Promise<void>>();
  const launcherRegistrations = new Map<string, PluginLauncherRegistration>();
  const dataHandlers = new Map<string, (params: Record<string, unknown>) => Promise<unknown>>();
  const actionHandlers = new Map<
    string,
    (params: Record<string, unknown>, context: PluginPerformActionContext) => Promise<unknown>
  >();
  const toolHandlers = new Map<string, {
    declaration: Pick<import("@paperclipai/shared").PluginToolDeclaration, "displayName" | "description" | "parametersSchema">;
    fn: (params: unknown, runCtx: ToolRunContext) => Promise<ToolResult>;
  }>();

  // Agent session event callbacks (populated by sendMessage, cleared by close)
  const sessionEventCallbacks = new Map<string, (event: AgentSessionEvent) => void>();

  // Pending outbound (worker→host) requests
  const pendingRequests = new Map<string | number, {
    resolve: (response: JsonRpcResponse) => void;
    timer: ReturnType<typeof setTimeout>;
  }>();
  let nextOutboundId = 1;
  const MAX_OUTBOUND_ID = Number.MAX_SAFE_INTEGER - 1;

  // -----------------------------------------------------------------------
  // Outbound messaging (worker → host)
  // -----------------------------------------------------------------------

  function sendMessage(message: unknown): void {
    if (!running) return;
    const serialized = serializeMessage(message as any);
    stdoutStream.write(serialized);
  }

  /**
   * Send a typed JSON-RPC request to the host and await the response.
   */
  function callHost<M extends WorkerToHostMethodName>(
    method: M,
    params: WorkerToHostMethods[M][0],
    timeoutMs?: number,
  ): Promise<WorkerToHostMethods[M][1]> {
    return new Promise<WorkerToHostMethods[M][1]>((resolve, reject) => {
      if (!running) {
        reject(new Error(`Cannot call "${method}" — worker RPC host is not running`));
        return;
      }

      if (nextOutboundId >= MAX_OUTBOUND_ID) {
        nextOutboundId = 1;
      }
      const id = nextOutboundId++;
      const timeout = timeoutMs ?? rpcTimeoutMs;
      let settled = false;

      const settle = <T>(fn: (value: T) => void, value: T): void => {
        if (settled) return;
        settled = true;
        clearTimeout(timer);
        pendingRequests.delete(id);
        fn(value);
      };

      const timer = setTimeout(() => {
        settle(
          reject,
          new JsonRpcCallError({
            code: PLUGIN_RPC_ERROR_CODES.TIMEOUT,
            message: `Worker→host call "${method}" timed out after ${timeout}ms`,
          }),
        );
      }, timeout);

      pendingRequests.set(id, {
        resolve: (response: JsonRpcResponse) => {
          if (isJsonRpcSuccessResponse(response)) {
            settle(resolve, response.result as WorkerToHostMethods[M][1]);
          } else if (isJsonRpcErrorResponse(response)) {
            settle(reject, new JsonRpcCallError(response.error));
          } else {
            settle(reject, new Error(`Unexpected response format for "${method}"`));
          }
        },
        timer,
      });

      try {
        const activeInvocation = invocationContextStorage.getStore();
        const request = {
          ...createRequest(method, params, id),
          ...(activeInvocation ? { paperclipInvocationId: activeInvocation.id } : {}),
        };
        sendMessage(request);
      } catch (err) {
        settle(reject, err instanceof Error ? err : new Error(String(err)));
      }
    });
  }

  /**
   * Send a JSON-RPC notification to the host (fire-and-forget).
   */
  function notifyHost(method: string, params: unknown): void {
    try {
      const activeInvocation = invocationContextStorage.getStore();
      sendMessage({
        ...createNotification(method, params),
        ...(activeInvocation ? { paperclipInvocationId: activeInvocation.id } : {}),
      });
    } catch {
      // Swallow — the host may have closed stdin
    }
  }

  // -----------------------------------------------------------------------
  // Build the PluginContext (SDK surface for plugin code)
  // -----------------------------------------------------------------------

  function buildContext(): PluginContext {
    return {
      get manifest() {
        if (!manifest) throw new Error("Plugin context accessed before initialization");
        return manifest;
      },

      config: {
        async get() {
          return callHost("config.get", {} as Record<string, never>);
        },
      },

      localFolders: {
        declarations() {
          if (!manifest) throw new Error("Plugin context accessed before initialization");
          return manifest.localFolders ?? [];
        },

        async configure(input) {
          return callHost("localFolders.configure", {
            companyId: input.companyId,
            folderKey: input.folderKey,
            path: input.path,
            access: input.access,
            requiredDirectories: input.requiredDirectories,
            requiredFiles: input.requiredFiles,
          });
        },

        async status(companyId: string, folderKey: string) {
          return callHost("localFolders.status", { companyId, folderKey });
        },

        async list(companyId: string, folderKey: string, options = {}) {
          return callHost("localFolders.list", {
            companyId,
            folderKey,
            relativePath: options.relativePath,
            recursive: options.recursive,
            maxEntries: options.maxEntries,
          });
        },

        async readText(companyId: string, folderKey: string, relativePath: string) {
          return callHost("localFolders.readText", { companyId, folderKey, relativePath });
        },

        async writeTextAtomic(companyId: string, folderKey: string, relativePath: string, contents: string) {
          return callHost("localFolders.writeTextAtomic", {
            companyId,
            folderKey,
            relativePath,
            contents,
          });
        },

        async deleteFile(companyId: string, folderKey: string, relativePath: string) {
          return callHost("localFolders.deleteFile", { companyId, folderKey, relativePath });
        },
      },

      events: {
        on(
          name: string,
          filterOrFn: EventFilter | ((event: PluginEvent) => Promise<void>),
          maybeFn?: (event: PluginEvent) => Promise<void>,
        ): () => void {
          let registration: EventRegistration;
          if (typeof filterOrFn === "function") {
            registration = { name, fn: filterOrFn };
          } else {
            if (!maybeFn) throw new Error("Event handler function is required");
            registration = { name, filter: filterOrFn, fn: maybeFn };
          }
          eventHandlers.push(registration);
          // Register subscription on the host so events are forwarded to this worker
          void callHost("events.subscribe", { eventPattern: name, filter: registration.filter ?? null }).catch((err) => {
            notifyHost("log", {
              level: "warn",
              message: `Failed to subscribe to event "${name}" on host: ${err instanceof Error ? err.message : String(err)}`,
            });
          });
          return () => {
            const idx = eventHandlers.indexOf(registration);
            if (idx !== -1) eventHandlers.splice(idx, 1);
          };
        },

        async emit(name: string, companyId: string, payload: unknown): Promise<void> {
          await callHost("events.emit", { name, companyId, payload });
        },
      },

      jobs: {
        register(key: string, fn: (job: PluginJobContext) => Promise<void>): void {
          jobHandlers.set(key, fn);
        },
      },

      launchers: {
        register(launcher: PluginLauncherRegistration): void {
          launcherRegistrations.set(launcher.id, launcher);
        },
      },

      db: {
        get namespace() {
          return databaseNamespace ?? "";
        },
        async query<T = Record<string, unknown>>(sql: string, params?: unknown[]): Promise<T[]> {
          return callHost("db.query", { sql, params }) as Promise<T[]>;
        },
        async execute(sql: string, params?: unknown[]) {
          return callHost("db.execute", { sql, params });
        },
      },

      http: {
        async fetch(url: string, init?: RequestInit): Promise<Response> {
          const serializedInit: Record<string, unknown> = {};
          if (init) {
            if (init.method) serializedInit.method = init.method;
            if (init.headers) {
              // Normalize headers to a plain object
              if (init.headers instanceof Headers) {
                const obj: Record<string, string> = {};
                init.headers.forEach((v, k) => { obj[k] = v; });
                serializedInit.headers = obj;
              } else if (Array.isArray(init.headers)) {
                const obj: Record<string, string> = {};
                for (const [k, v] of init.headers) obj[k] = v;
                serializedInit.headers = obj;
              } else {
                serializedInit.headers = init.headers;
              }
            }
            if (init.body !== undefined && init.body !== null) {
              serializedInit.body = typeof init.body === "string"
                ? init.body
                : String(init.body);
            }
          }

          const result = await callHost("http.fetch", {
            url,
            init: Object.keys(serializedInit).length > 0 ? serializedInit : undefined,
          });

          // Reconstruct a Response-like object from the serialized result
          return new Response(result.body, {
            status: result.status,
            statusText: result.statusText,
            headers: result.headers,
          });
        },
      },

      secrets: {
        async resolve(secretRef: string): Promise<string> {
          return callHost("secrets.resolve", { secretRef });
        },
      },

      activity: {
        async log(entry): Promise<void> {
          await callHost("activity.log", {
            companyId: entry.companyId,
            message: entry.message,
            entityType: entry.entityType,
            entityId: entry.entityId,
            metadata: entry.metadata,
          });
        },
      },

      state: {
        async get(input: ScopeKey): Promise<unknown> {
          return callHost("state.get", {
            scopeKind: input.scopeKind,
            scopeId: input.scopeId,
            namespace: input.namespace,
            stateKey: input.stateKey,
          });
        },

        async set(input: ScopeKey, value: unknown): Promise<void> {
          await callHost("state.set", {
            scopeKind: input.scopeKind,
            scopeId: input.scopeId,
            namespace: input.namespace,
            stateKey: input.stateKey,
            value,
          });
        },

        async delete(input: ScopeKey): Promise<void> {
          await callHost("state.delete", {
            scopeKind: input.scopeKind,
            scopeId: input.scopeId,
            namespace: input.namespace,
            stateKey: input.stateKey,
          });
        },
      },

      entities: {
        async upsert(input) {
          return callHost("entities.upsert", {
            entityType: input.entityType,
            scopeKind: input.scopeKind,
            scopeId: input.scopeId,
            externalId: input.externalId,
            title: input.title,
            status: input.status,
            data: input.data,
          });
        },

        async list(query) {
          return callHost("entities.list", {
            entityType: query.entityType,
            scopeKind: query.scopeKind,
            scopeId: query.scopeId,
            externalId: query.externalId,
            limit: query.limit,
            offset: query.offset,
          });
        },
      },

      projects: {
        async list(input) {
          return callHost("projects.list", {
            companyId: input.companyId,
            limit: input.limit,
            offset: input.offset,
          });
        },

        async get(projectId: string, companyId: string) {
          return callHost("projects.get", { projectId, companyId });
        },

        async listWorkspaces(projectId: string, companyId: string) {
          return callHost("projects.listWorkspaces", { projectId, companyId });
        },

        async getPrimaryWorkspace(projectId: string, companyId: string) {
          return callHost("projects.getPrimaryWorkspace", { projectId, companyId });
        },

        async getWorkspaceForIssue(issueId: string, companyId: string) {
          return callHost("projects.getWorkspaceForIssue", { issueId, companyId });
        },

        managed: {
          async get(projectKey: string, companyId: string) {
            return callHost("projects.managed.get", { projectKey, companyId });
          },
          async reconcile(projectKey: string, companyId: string) {
            return callHost("projects.managed.reconcile", { projectKey, companyId });
          },
          async reset(projectKey: string, companyId: string) {
            return callHost("projects.managed.reset", { projectKey, companyId });
          },
        },
      },

      executionWorkspaces: {
        async get(workspaceId: string, companyId: string) {
          return callHost("executionWorkspaces.get", { workspaceId, companyId });
        },
      },

      routines: {
        managed: {
          async get(routineKey: string, companyId: string) {
            return callHost("routines.managed.get", { routineKey, companyId });
          },
          async reconcile(
            routineKey: string,
            companyId: string,
            overrides?: { assigneeAgentId?: string | null; projectId?: string | null },
          ) {
            return callHost("routines.managed.reconcile", { routineKey, companyId, ...overrides });
          },
          async reset(
            routineKey: string,
            companyId: string,
            overrides?: { assigneeAgentId?: string | null; projectId?: string | null },
          ) {
            return callHost("routines.managed.reset", { routineKey, companyId, ...overrides });
          },
          async update(routineKey: string, companyId: string, patch: { status?: string }) {
            return callHost("routines.managed.update", { routineKey, companyId, ...patch });
          },
          async run(
            routineKey: string,
            companyId: string,
            overrides?: { assigneeAgentId?: string | null; projectId?: string | null },
          ) {
            return callHost("routines.managed.run", { routineKey, companyId, ...overrides });
          },
        },
      },

      skills: {
        managed: {
          async get(skillKey: string, companyId: string) {
            return callHost("skills.managed.get", { skillKey, companyId });
          },
          async reconcile(skillKey: string, companyId: string) {
            return callHost("skills.managed.reconcile", { skillKey, companyId });
          },
          async reset(skillKey: string, companyId: string) {
            return callHost("skills.managed.reset", { skillKey, companyId });
          },
        },
      },

      companies: {
        async list(input) {
          return callHost("companies.list", {
            limit: input?.limit,
            offset: input?.offset,
          });
        },

        async get(companyId: string) {
          return callHost("companies.get", { companyId });
        },
      },

      issues: {
        async list(input) {
          return callHost("issues.list", {
            companyId: input.companyId,
            projectId: input.projectId,
            assigneeAgentId: input.assigneeAgentId,
            originKind: input.originKind,
            originKindPrefix: input.originKindPrefix,
            originId: input.originId,
            status: input.status,
            includePluginOperations: input.includePluginOperations,
            limit: input.limit,
            offset: input.offset,
          });
        },

        async get(issueId: string, companyId: string) {
          return callHost("issues.get", { issueId, companyId });
        },

        async create(input) {
          return callHost("issues.create", {
            companyId: input.companyId,
            projectId: input.projectId,
            goalId: input.goalId,
            parentId: input.parentId,
            inheritExecutionWorkspaceFromIssueId: input.inheritExecutionWorkspaceFromIssueId,
            title: input.title,
            description: input.description,
            status: input.status,
            priority: input.priority,
            assigneeAgentId: input.assigneeAgentId,
            assigneeUserId: input.assigneeUserId,
            requestDepth: input.requestDepth,
            billingCode: input.billingCode,
            assigneeAdapterOverrides: input.assigneeAdapterOverrides,
            surfaceVisibility: input.surfaceVisibility,
            originKind: input.originKind,
            originId: input.originId,
            originRunId: input.originRunId,
            blockedByIssueIds: input.blockedByIssueIds,
            labelIds: input.labelIds,
            executionWorkspaceId: input.executionWorkspaceId,
            executionWorkspacePreference: input.executionWorkspacePreference,
            executionWorkspaceSettings: input.executionWorkspaceSettings,
            actorAgentId: input.actor?.actorAgentId,
            actorUserId: input.actor?.actorUserId,
            actorRunId: input.actor?.actorRunId,
          });
        },

        async update(issueId: string, patch, companyId: string, actor) {
          return callHost("issues.update", {
            issueId,
            patch: {
              ...(patch as Record<string, unknown>),
              actorAgentId: actor?.actorAgentId,
              actorUserId: actor?.actorUserId,
              actorRunId: actor?.actorRunId,
            },
            companyId,
          });
        },

        async assertCheckoutOwner(input) {
          return callHost("issues.assertCheckoutOwner", input);
        },

        async getSubtree(issueId: string, companyId: string, options) {
          return callHost("issues.getSubtree", {
            issueId,
            companyId,
            includeRoot: options?.includeRoot,
            includeRelations: options?.includeRelations,
            includeDocuments: options?.includeDocuments,
            includeActiveRuns: options?.includeActiveRuns,
            includeAssignees: options?.includeAssignees,
          });
        },

        async requestWakeup(issueId: string, companyId: string, options) {
          return callHost("issues.requestWakeup", {
            issueId,
            companyId,
            reason: options?.reason,
            contextSource: options?.contextSource,
            idempotencyKey: options?.idempotencyKey,
            actorAgentId: options?.actorAgentId,
            actorUserId: options?.actorUserId,
            actorRunId: options?.actorRunId,
          });
        },

        async requestWakeups(issueIds: string[], companyId: string, options) {
          return callHost("issues.requestWakeups", {
            issueIds,
            companyId,
            reason: options?.reason,
            contextSource: options?.contextSource,
            idempotencyKeyPrefix: options?.idempotencyKeyPrefix,
            actorAgentId: options?.actorAgentId,
            actorUserId: options?.actorUserId,
            actorRunId: options?.actorRunId,
          });
        },

        async listComments(issueId: string, companyId: string) {
          return callHost("issues.listComments", { issueId, companyId });
        },

        async createComment(issueId: string, body: string, companyId: string, options?: { authorAgentId?: string }) {
          return callHost("issues.createComment", { issueId, body, companyId, authorAgentId: options?.authorAgentId });
        },

        async createInteraction(issueId: string, interaction, companyId: string, options?: { authorAgentId?: string }) {
          return callHost("issues.createInteraction", {
            issueId,
            companyId,
            interaction,
            authorAgentId: options?.authorAgentId,
          });
        },

        async suggestTasks(
          issueId: string,
          interaction,
          companyId: string,
          options?: { authorAgentId?: string },
        ): Promise<SuggestTasksInteraction> {
          return callHost("issues.createInteraction", {
            issueId,
            companyId,
            interaction: {
              ...interaction,
              kind: "suggest_tasks",
            },
            authorAgentId: options?.authorAgentId,
          }) as Promise<SuggestTasksInteraction>;
        },

        async askUserQuestions(
          issueId: string,
          interaction,
          companyId: string,
          options?: { authorAgentId?: string },
        ): Promise<AskUserQuestionsInteraction> {
          return callHost("issues.createInteraction", {
            issueId,
            companyId,
            interaction: {
              ...interaction,
              kind: "ask_user_questions",
            },
            authorAgentId: options?.authorAgentId,
          }) as Promise<AskUserQuestionsInteraction>;
        },

        async requestConfirmation(
          issueId: string,
          interaction,
          companyId: string,
          options?: { authorAgentId?: string },
        ): Promise<RequestConfirmationInteraction> {
          return callHost("issues.createInteraction", {
            issueId,
            companyId,
            interaction: {
              ...interaction,
              kind: "request_confirmation",
            },
            authorAgentId: options?.authorAgentId,
          }) as Promise<RequestConfirmationInteraction>;
        },

        async requestCheckboxConfirmation(
          issueId: string,
          interaction,
          companyId: string,
          options?: { authorAgentId?: string },
        ): Promise<RequestCheckboxConfirmationInteraction> {
          return callHost("issues.createInteraction", {
            issueId,
            companyId,
            interaction: {
              ...interaction,
              kind: "request_checkbox_confirmation",
            },
            authorAgentId: options?.authorAgentId,
          }) as Promise<RequestCheckboxConfirmationInteraction>;
        },

        documents: {
          async list(issueId: string, companyId: string) {
            return callHost("issues.documents.list", { issueId, companyId });
          },

          async get(issueId: string, key: string, companyId: string) {
            return callHost("issues.documents.get", { issueId, key, companyId });
          },

          async upsert(input) {
            return callHost("issues.documents.upsert", {
              issueId: input.issueId,
              key: input.key,
              body: input.body,
              companyId: input.companyId,
              title: input.title,
              format: input.format,
              changeSummary: input.changeSummary,
            });
          },

          async delete(issueId: string, key: string, companyId: string) {
            return callHost("issues.documents.delete", { issueId, key, companyId });
          },
        },

        relations: {
          async get(issueId: string, companyId: string) {
            return callHost("issues.relations.get", { issueId, companyId });
          },

          async setBlockedBy(issueId: string, blockedByIssueIds: string[], companyId: string, actor) {
            return callHost("issues.relations.setBlockedBy", {
              issueId,
              companyId,
              blockedByIssueIds,
              actorAgentId: actor?.actorAgentId,
              actorUserId: actor?.actorUserId,
              actorRunId: actor?.actorRunId,
            });
          },

          async addBlockers(issueId: string, blockerIssueIds: string[], companyId: string, actor) {
            return callHost("issues.relations.addBlockers", {
              issueId,
              companyId,
              blockerIssueIds,
              actorAgentId: actor?.actorAgentId,
              actorUserId: actor?.actorUserId,
              actorRunId: actor?.actorRunId,
            });
          },

          async removeBlockers(issueId: string, blockerIssueIds: string[], companyId: string, actor) {
            return callHost("issues.relations.removeBlockers", {
              issueId,
              companyId,
              blockerIssueIds,
              actorAgentId: actor?.actorAgentId,
              actorUserId: actor?.actorUserId,
              actorRunId: actor?.actorRunId,
            });
          },
        },

        summaries: {
          async getOrchestration(input) {
            return callHost("issues.summaries.getOrchestration", input);
          },
        },
      },

      agents: {
        async list(input) {
          return callHost("agents.list", {
            companyId: input.companyId,
            status: input.status,
            limit: input.limit,
            offset: input.offset,
          });
        },

        async get(agentId: string, companyId: string) {
          return callHost("agents.get", { agentId, companyId });
        },

        async pause(agentId: string, companyId: string) {
          return callHost("agents.pause", { agentId, companyId });
        },

        async resume(agentId: string, companyId: string) {
          return callHost("agents.resume", { agentId, companyId });
        },

        async invoke(agentId: string, companyId: string, opts: { prompt: string; reason?: string }) {
          return callHost("agents.invoke", { agentId, companyId, prompt: opts.prompt, reason: opts.reason });
        },

        managed: {
          async get(agentKey: string, companyId: string) {
            return callHost("agents.managed.get", { agentKey, companyId });
          },

          async reconcile(agentKey: string, companyId: string) {
            return callHost("agents.managed.reconcile", { agentKey, companyId });
          },

          async reset(agentKey: string, companyId: string) {
            return callHost("agents.managed.reset", { agentKey, companyId });
          },
        },

        sessions: {
          async create(agentId: string, companyId: string, opts?: { taskKey?: string; reason?: string }) {
            return callHost("agents.sessions.create", {
              agentId,
              companyId,
              taskKey: opts?.taskKey,
              reason: opts?.reason,
            });
          },

          async list(agentId: string, companyId: string) {
            return callHost("agents.sessions.list", { agentId, companyId });
          },

          async sendMessage(sessionId: string, companyId: string, opts: {
            prompt: string;
            reason?: string;
            onEvent?: (event: AgentSessionEvent) => void;
          }) {
            if (opts.onEvent) {
              sessionEventCallbacks.set(sessionId, opts.onEvent);
            }
            try {
              return await callHost("agents.sessions.sendMessage", {
                sessionId,
                companyId,
                prompt: opts.prompt,
                reason: opts.reason,
              });
            } catch (err) {
              sessionEventCallbacks.delete(sessionId);
              throw err;
            }
          },

          async close(sessionId: string, companyId: string) {
            sessionEventCallbacks.delete(sessionId);
            await callHost("agents.sessions.close", { sessionId, companyId });
          },
        },
      },

      goals: {
        async list(input) {
          return callHost("goals.list", {
            companyId: input.companyId,
            level: input.level,
            status: input.status,
            limit: input.limit,
            offset: input.offset,
          });
        },

        async get(goalId: string, companyId: string) {
          return callHost("goals.get", { goalId, companyId });
        },

        async create(input) {
          return callHost("goals.create", {
            companyId: input.companyId,
            title: input.title,
            description: input.description,
            level: input.level,
            status: input.status,
            parentId: input.parentId,
            ownerAgentId: input.ownerAgentId,
          });
        },

        async update(goalId: string, patch, companyId: string) {
          return callHost("goals.update", {
            goalId,
            patch: patch as Record<string, unknown>,
            companyId,
          });
        },
      },

      access: {
        members: {
          async list(input) {
            return callHost("access.members.list", {
              companyId: input.companyId,
              includeArchived: input.includeArchived,
            });
          },

          async get(memberId: string, companyId: string) {
            return callHost("access.members.get", { memberId, companyId });
          },

          async update(memberId: string, patch, companyId: string) {
            return callHost("access.members.update", { memberId, patch, companyId });
          },
        },

        invites: {
          async list(input) {
            return callHost("access.invites.list", {
              companyId: input.companyId,
              state: input.state,
              limit: input.limit,
              offset: input.offset,
            });
          },

          async create(input) {
            return callHost("access.invites.create", {
              companyId: input.companyId,
              allowedJoinTypes: input.allowedJoinTypes,
              humanRole: input.humanRole,
              defaultsPayload: input.defaultsPayload,
              agentMessage: input.agentMessage,
            });
          },

          async revoke(inviteId: string, companyId: string) {
            return callHost("access.invites.revoke", { inviteId, companyId });
          },
        },
      },

      authorization: {
        grants: {
          async list(input) {
            return callHost("authorization.grants.list", input);
          },
          async set(input) {
            return callHost("authorization.grants.set", input);
          },
        },

        policies: {
          async summary(companyId: string) {
            return callHost("authorization.policies.summary", { companyId });
          },
          async get(input) {
            return callHost("authorization.policies.get", input);
          },
          async update(input) {
            return callHost("authorization.policies.update", input);
          },
          async previewAssignment(input) {
            return callHost("authorization.policies.previewAssignment", input);
          },
          async explainAssignment(input) {
            return callHost("authorization.policies.explainAssignment", input);
          },
        },

        audit: {
          async search(input) {
            return callHost("authorization.audit.search", input);
          },
        },
      },

      data: {
        register(key: string, handler: (params: Record<string, unknown>) => Promise<unknown>): void {
          dataHandlers.set(key, handler);
        },
      },

      actions: {
        register(
          key: string,
          handler: (params: Record<string, unknown>, context: PluginPerformActionContext) => Promise<unknown>,
        ): void {
          actionHandlers.set(key, handler);
        },
      },

      streams: (() => {
        // Track channel → companyId so emit/close don't require companyId
        const channelCompanyMap = new Map<string, string>();
        return {
          open(channel: string, companyId: string): void {
            channelCompanyMap.set(channel, companyId);
            notifyHost("streams.open", { channel, companyId });
          },
          emit(channel: string, event: unknown): void {
            const companyId = channelCompanyMap.get(channel) ?? "";
            notifyHost("streams.emit", { channel, companyId, event });
          },
          close(channel: string): void {
            const companyId = channelCompanyMap.get(channel) ?? "";
            channelCompanyMap.delete(channel);
            notifyHost("streams.close", { channel, companyId });
          },
        };
      })(),

      tools: {
        register(
          name: string,
          declaration: Pick<import("@paperclipai/shared").PluginToolDeclaration, "displayName" | "description" | "parametersSchema">,
          fn: (params: unknown, runCtx: ToolRunContext) => Promise<ToolResult>,
        ): void {
          toolHandlers.set(name, { declaration, fn });
        },
      },

      metrics: {
        async write(name: string, value: number, tags?: Record<string, string>): Promise<void> {
          await callHost("metrics.write", { name, value, tags });
        },
      },

      telemetry: {
        async track(
          eventName: string,
          dimensions?: Record<string, string | number | boolean>,
        ): Promise<void> {
          await callHost("telemetry.track", { eventName, dimensions });
        },
      },

      logger: {
        info(message: string, meta?: Record<string, unknown>): void {
          notifyHost("log", { level: "info", message, meta });
        },
        warn(message: string, meta?: Record<string, unknown>): void {
          notifyHost("log", { level: "warn", message, meta });
        },
        error(message: string, meta?: Record<string, unknown>): void {
          notifyHost("log", { level: "error", message, meta });
        },
        debug(message: string, meta?: Record<string, unknown>): void {
          notifyHost("log", { level: "debug", message, meta });
        },
      },
    };
  }

  const ctx = buildContext();

  // -----------------------------------------------------------------------
  // Inbound message handling (host → worker)
  // -----------------------------------------------------------------------

  /**
   * Handle an incoming JSON-RPC request from the host.
   *
   * Dispatches to the correct handler based on the method name.
   */
  async function handleHostRequest(request: JsonRpcRequest): Promise<void> {
    const { id, method, params } = request;

    try {
      const invoke = () => dispatchMethod(method, params);
      const result = request.paperclipInvocation
        ? await invocationContextStorage.run(request.paperclipInvocation, invoke)
        : await invoke();
      sendMessage(createSuccessResponse(id, result ?? null));
    } catch (err) {
      const errorMessage = err instanceof Error ? err.message : String(err);
      // Propagate specific error codes from handler errors (e.g.
      // METHOD_NOT_FOUND, METHOD_NOT_IMPLEMENTED) — fall back to
      // WORKER_ERROR for untyped exceptions.
      const errorCode =
        typeof (err as any)?.code === "number"
          ? (err as any).code
          : PLUGIN_RPC_ERROR_CODES.WORKER_ERROR;

      sendMessage(createErrorResponse(id, errorCode, errorMessage));
    }
  }

  /**
   * Dispatch a host→worker method call to the appropriate handler.
   */
  async function dispatchMethod(method: string, params: unknown): Promise<unknown> {
    switch (method) {
      case "initialize":
        return handleInitialize(params as InitializeParams);

      case "health":
        return handleHealth();

      case "shutdown":
        return handleShutdown();

      case "validateConfig":
        return handleValidateConfig(params as ValidateConfigParams);

      case "configChanged":
        return handleConfigChanged(params as ConfigChangedParams);

      case "onEvent":
        return handleOnEvent(params as OnEventParams);

      case "runJob":
        return handleRunJob(params as RunJobParams);

      case "handleWebhook":
        return handleWebhook(params as PluginWebhookInput);

      case "handleApiRequest":
        return handleApiRequest(params as PluginApiRequestInput);

      case "getData":
        return handleGetData(params as GetDataParams);

      case "performAction":
        return handlePerformAction(params as PerformActionParams);

      case "executeTool":
        return handleExecuteTool(params as ExecuteToolParams);
      case "detectExternalObjects":
        return handleDetectExternalObjects(params as DetectExternalObjectsParams);
      case "resolveExternalObject":
        return handleResolveExternalObject(params as ResolveExternalObjectParams);
      case "refreshExternalObjects":
        return handleRefreshExternalObjects(params as RefreshExternalObjectsParams);

      case "environmentValidateConfig":
        return handleEnvironmentValidateConfig(params as PluginEnvironmentValidateConfigParams);

      case "environmentProbe":
        return handleEnvironmentProbe(params as PluginEnvironmentProbeParams);

      case "environmentAcquireLease":
        return handleEnvironmentAcquireLease(params as PluginEnvironmentAcquireLeaseParams);

      case "environmentResumeLease":
        return handleEnvironmentResumeLease(params as PluginEnvironmentResumeLeaseParams);

      case "environmentReleaseLease":
        return handleEnvironmentReleaseLease(params as PluginEnvironmentReleaseLeaseParams);

      case "environmentDestroyLease":
        return handleEnvironmentDestroyLease(params as PluginEnvironmentDestroyLeaseParams);

      case "environmentRealizeWorkspace":
        return handleEnvironmentRealizeWorkspace(params as PluginEnvironmentRealizeWorkspaceParams);

      case "environmentExecute":
        return handleEnvironmentExecute(params as PluginEnvironmentExecuteParams);

      case "environmentStartInteractiveSetup":
        return handleEnvironmentStartInteractiveSetup(params as PluginEnvironmentStartInteractiveSetupParams);

      case "environmentGetInteractiveSetup":
        return handleEnvironmentGetInteractiveSetup(params as PluginEnvironmentGetInteractiveSetupParams);

      case "environmentCaptureTemplate":
        return handleEnvironmentCaptureTemplate(params as PluginEnvironmentCaptureTemplateParams);

      case "environmentCancelInteractiveSetup":
        return handleEnvironmentCancelInteractiveSetup(params as PluginEnvironmentCancelInteractiveSetupParams);

      case "environmentDeleteTemplate":
        return handleEnvironmentDeleteTemplate(params as PluginEnvironmentDeleteTemplateParams);

      default:
        throw Object.assign(
          new Error(`Unknown method: ${method}`),
          { code: JSONRPC_ERROR_CODES.METHOD_NOT_FOUND },
        );
    }
  }

  // -----------------------------------------------------------------------
  // Host→Worker method handlers
  // -----------------------------------------------------------------------

  async function handleInitialize(params: InitializeParams): Promise<InitializeResult> {
    if (initialized) {
      throw new Error("Worker already initialized");
    }

    manifest = params.manifest;
    currentConfig = params.config;
    databaseNamespace = params.databaseNamespace ?? null;

    // Call the plugin's setup function
    await plugin.definition.setup(ctx);

    initialized = true;

    // Report which optional methods this plugin implements
    const supportedMethods: string[] = [];
    if (plugin.definition.onValidateConfig) supportedMethods.push("validateConfig");
    if (plugin.definition.onConfigChanged) supportedMethods.push("configChanged");
    if (plugin.definition.onHealth) supportedMethods.push("health");
    if (plugin.definition.onShutdown) supportedMethods.push("shutdown");
    if (plugin.definition.onApiRequest) supportedMethods.push("handleApiRequest");
    if (plugin.definition.onDetectExternalObjects) supportedMethods.push("detectExternalObjects");
    if (plugin.definition.onResolveExternalObject) supportedMethods.push("resolveExternalObject");
    if (plugin.definition.onRefreshExternalObjects) supportedMethods.push("refreshExternalObjects");
    if (plugin.definition.onEnvironmentValidateConfig) supportedMethods.push("environmentValidateConfig");
    if (plugin.definition.onEnvironmentProbe) supportedMethods.push("environmentProbe");
    if (plugin.definition.onEnvironmentAcquireLease) supportedMethods.push("environmentAcquireLease");
    if (plugin.definition.onEnvironmentResumeLease) supportedMethods.push("environmentResumeLease");
    if (plugin.definition.onEnvironmentReleaseLease) supportedMethods.push("environmentReleaseLease");
    if (plugin.definition.onEnvironmentDestroyLease) supportedMethods.push("environmentDestroyLease");
    if (plugin.definition.onEnvironmentRealizeWorkspace) supportedMethods.push("environmentRealizeWorkspace");
    if (plugin.definition.onEnvironmentExecute) supportedMethods.push("environmentExecute");
    if (plugin.definition.onEnvironmentStartInteractiveSetup) supportedMethods.push("environmentStartInteractiveSetup");
    if (plugin.definition.onEnvironmentGetInteractiveSetup) supportedMethods.push("environmentGetInteractiveSetup");
    if (plugin.definition.onEnvironmentCaptureTemplate) supportedMethods.push("environmentCaptureTemplate");
    if (plugin.definition.onEnvironmentCancelInteractiveSetup) supportedMethods.push("environmentCancelInteractiveSetup");
    if (plugin.definition.onEnvironmentDeleteTemplate) supportedMethods.push("environmentDeleteTemplate");

    return { ok: true, supportedMethods };
  }

  async function handleHealth(): Promise<PluginHealthDiagnostics> {
    if (plugin.definition.onHealth) {
      return plugin.definition.onHealth();
    }
    // Default: report OK if the worker is alive
    return { status: "ok" };
  }

  async function handleShutdown(): Promise<void> {
    if (plugin.definition.onShutdown) {
      await plugin.definition.onShutdown();
    }

    // Schedule cleanup after we send the response.
    // Use setImmediate to let the response flush before exiting.
    // Only call process.exit() when running with real process streams.
    // When custom streams are provided (tests), just clean up.
    setImmediate(() => {
      cleanup();
      if (!options.stdin && !options.stdout) {
        process.exit(0);
      }
    });
  }

  async function handleValidateConfig(
    params: ValidateConfigParams,
  ): Promise<PluginConfigValidationResult> {
    if (!plugin.definition.onValidateConfig) {
      throw Object.assign(
        new Error("validateConfig is not implemented by this plugin"),
        { code: PLUGIN_RPC_ERROR_CODES.METHOD_NOT_IMPLEMENTED },
      );
    }
    return plugin.definition.onValidateConfig(params.config);
  }

  async function handleConfigChanged(params: ConfigChangedParams): Promise<void> {
    currentConfig = params.config;

    if (plugin.definition.onConfigChanged) {
      await plugin.definition.onConfigChanged(params.config);
    }
  }

  async function handleOnEvent(params: OnEventParams): Promise<void> {
    const event = params.event;

    for (const registration of eventHandlers) {
      // Check event type match
      const exactMatch = registration.name === event.eventType;
      const wildcardPluginAll =
        registration.name === "plugin.*" &&
        event.eventType.startsWith("plugin.");
      const wildcardPluginOne =
        registration.name.endsWith(".*") &&
        event.eventType.startsWith(registration.name.slice(0, -1));

      if (!exactMatch && !wildcardPluginAll && !wildcardPluginOne) continue;

      // Check filter
      if (registration.filter && !allowsEvent(registration.filter, event)) continue;

      try {
        await registration.fn(event);
      } catch (err) {
        // Log error but continue processing other handlers so one failing
        // handler doesn't prevent the rest from running.
        notifyHost("log", {
          level: "error",
          message: `Event handler for "${registration.name}" failed: ${
            err instanceof Error ? err.message : String(err)
          }`,
          meta: { eventType: event.eventType, stack: err instanceof Error ? err.stack : undefined },
        });
      }
    }
  }

  async function handleRunJob(params: RunJobParams): Promise<void> {
    const handler = jobHandlers.get(params.job.jobKey);
    if (!handler) {
      throw new Error(`No handler registered for job "${params.job.jobKey}"`);
    }
    await handler(params.job);
  }

  async function handleWebhook(params: PluginWebhookInput): Promise<void> {
    if (!plugin.definition.onWebhook) {
      throw Object.assign(
        new Error("handleWebhook is not implemented by this plugin"),
        { code: PLUGIN_RPC_ERROR_CODES.METHOD_NOT_IMPLEMENTED },
      );
    }
    await plugin.definition.onWebhook(params);
  }

  async function handleApiRequest(params: PluginApiRequestInput): Promise<unknown> {
    if (!plugin.definition.onApiRequest) {
      throw Object.assign(
        new Error("handleApiRequest is not implemented by this plugin"),
        { code: PLUGIN_RPC_ERROR_CODES.METHOD_NOT_IMPLEMENTED },
      );
    }
    return plugin.definition.onApiRequest(params);
  }

  async function handleGetData(params: GetDataParams): Promise<unknown> {
    const handler = dataHandlers.get(params.key);
    if (!handler) {
      throw new Error(`No data handler registered for key "${params.key}"`);
    }
    return handler({
      ...params.params,
      ...(params.companyId === undefined ? {} : { companyId: params.companyId }),
      ...(params.renderEnvironment === undefined ? {} : { renderEnvironment: params.renderEnvironment }),
    });
  }

  function stringOrNull(value: unknown): string | null {
    return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
  }

  function actorTypeOrSystem(value: unknown): PluginPerformActionActorContext["type"] {
    return value === "user" || value === "agent" || value === "system" ? value : "system";
  }

  function actionContextFromParams(params: PerformActionParams): PluginPerformActionContext {
    const rawActor = params.actorContext && typeof params.actorContext === "object"
      ? params.actorContext
      : null;
    const actor = Object.freeze({
      type: actorTypeOrSystem(rawActor?.type),
      userId: stringOrNull(rawActor?.userId),
      agentId: stringOrNull(rawActor?.agentId),
      runId: stringOrNull(rawActor?.runId),
      companyId: stringOrNull(rawActor?.companyId),
    });
    return Object.freeze({
      actor,
      companyId: actor.companyId,
    });
  }

  async function handlePerformAction(params: PerformActionParams): Promise<unknown> {
    const handler = actionHandlers.get(params.key);
    if (!handler) {
      throw new Error(`No action handler registered for key "${params.key}"`);
    }
    return handler(
      {
        ...params.params,
        ...(params.companyId === undefined ? {} : { companyId: params.companyId }),
        ...(params.renderEnvironment === undefined ? {} : { renderEnvironment: params.renderEnvironment }),
      },
      actionContextFromParams(params),
    );
  }

  async function handleExecuteTool(params: ExecuteToolParams): Promise<ToolResult> {
    const entry = toolHandlers.get(params.toolName);
    if (!entry) {
      throw new Error(`No tool handler registered for "${params.toolName}"`);
    }
    return entry.fn(params.parameters, params.runContext);
  }

  async function handleDetectExternalObjects(params: DetectExternalObjectsParams) {
    if (!plugin.definition.onDetectExternalObjects) {
      throw methodNotImplemented("detectExternalObjects");
    }
    return plugin.definition.onDetectExternalObjects(params);
  }

  async function handleResolveExternalObject(params: ResolveExternalObjectParams) {
    if (!plugin.definition.onResolveExternalObject) {
      throw methodNotImplemented("resolveExternalObject");
    }
    return plugin.definition.onResolveExternalObject(params);
  }

  async function handleRefreshExternalObjects(params: RefreshExternalObjectsParams) {
    if (!plugin.definition.onRefreshExternalObjects) {
      throw methodNotImplemented("refreshExternalObjects");
    }
    return plugin.definition.onRefreshExternalObjects(params);
  }

  function methodNotImplemented(method: string): Error & { code: number } {
    return Object.assign(
      new Error(`${method} is not implemented by this plugin`),
      { code: PLUGIN_RPC_ERROR_CODES.METHOD_NOT_IMPLEMENTED },
    );
  }

  async function handleEnvironmentValidateConfig(
    params: PluginEnvironmentValidateConfigParams,
  ) {
    if (!plugin.definition.onEnvironmentValidateConfig) {
      throw methodNotImplemented("environmentValidateConfig");
    }
    return plugin.definition.onEnvironmentValidateConfig(params);
  }

  async function handleEnvironmentProbe(params: PluginEnvironmentProbeParams) {
    if (!plugin.definition.onEnvironmentProbe) {
      throw methodNotImplemented("environmentProbe");
    }
    return plugin.definition.onEnvironmentProbe(params);
  }

  async function handleEnvironmentAcquireLease(params: PluginEnvironmentAcquireLeaseParams) {
    if (!plugin.definition.onEnvironmentAcquireLease) {
      throw methodNotImplemented("environmentAcquireLease");
    }
    return plugin.definition.onEnvironmentAcquireLease(params);
  }

  async function handleEnvironmentResumeLease(params: PluginEnvironmentResumeLeaseParams) {
    if (!plugin.definition.onEnvironmentResumeLease) {
      throw methodNotImplemented("environmentResumeLease");
    }
    return plugin.definition.onEnvironmentResumeLease(params);
  }

  async function handleEnvironmentReleaseLease(params: PluginEnvironmentReleaseLeaseParams) {
    if (!plugin.definition.onEnvironmentReleaseLease) {
      throw methodNotImplemented("environmentReleaseLease");
    }
    return plugin.definition.onEnvironmentReleaseLease(params);
  }

  async function handleEnvironmentDestroyLease(params: PluginEnvironmentDestroyLeaseParams) {
    if (!plugin.definition.onEnvironmentDestroyLease) {
      throw methodNotImplemented("environmentDestroyLease");
    }
    return plugin.definition.onEnvironmentDestroyLease(params);
  }

  async function handleEnvironmentRealizeWorkspace(params: PluginEnvironmentRealizeWorkspaceParams) {
    if (!plugin.definition.onEnvironmentRealizeWorkspace) {
      throw methodNotImplemented("environmentRealizeWorkspace");
    }
    return plugin.definition.onEnvironmentRealizeWorkspace(params);
  }

  async function handleEnvironmentExecute(params: PluginEnvironmentExecuteParams) {
    if (!plugin.definition.onEnvironmentExecute) {
      throw methodNotImplemented("environmentExecute");
    }
    return plugin.definition.onEnvironmentExecute(params);
  }

  async function handleEnvironmentStartInteractiveSetup(params: PluginEnvironmentStartInteractiveSetupParams) {
    if (!plugin.definition.onEnvironmentStartInteractiveSetup) {
      throw methodNotImplemented("environmentStartInteractiveSetup");
    }
    return plugin.definition.onEnvironmentStartInteractiveSetup(params);
  }

  async function handleEnvironmentGetInteractiveSetup(params: PluginEnvironmentGetInteractiveSetupParams) {
    if (!plugin.definition.onEnvironmentGetInteractiveSetup) {
      throw methodNotImplemented("environmentGetInteractiveSetup");
    }
    return plugin.definition.onEnvironmentGetInteractiveSetup(params);
  }

  async function handleEnvironmentCaptureTemplate(params: PluginEnvironmentCaptureTemplateParams) {
    if (!plugin.definition.onEnvironmentCaptureTemplate) {
      throw methodNotImplemented("environmentCaptureTemplate");
    }
    return plugin.definition.onEnvironmentCaptureTemplate(params);
  }

  async function handleEnvironmentCancelInteractiveSetup(params: PluginEnvironmentCancelInteractiveSetupParams) {
    if (!plugin.definition.onEnvironmentCancelInteractiveSetup) {
      throw methodNotImplemented("environmentCancelInteractiveSetup");
    }
    return plugin.definition.onEnvironmentCancelInteractiveSetup(params);
  }

  async function handleEnvironmentDeleteTemplate(params: PluginEnvironmentDeleteTemplateParams) {
    if (!plugin.definition.onEnvironmentDeleteTemplate) {
      throw methodNotImplemented("environmentDeleteTemplate");
    }
    return plugin.definition.onEnvironmentDeleteTemplate(params);
  }

  // -----------------------------------------------------------------------
  // Event filter helper
  // -----------------------------------------------------------------------

  function allowsEvent(filter: EventFilter, event: PluginEvent): boolean {
    const payload = event.payload as Record<string, unknown> | undefined;

    if (filter.companyId !== undefined) {
      const companyId = event.companyId ?? String(payload?.companyId ?? "");
      if (companyId !== filter.companyId) return false;
    }

    if (filter.projectId !== undefined) {
      const projectId = event.entityType === "project"
        ? event.entityId
        : String(payload?.projectId ?? "");
      if (projectId !== filter.projectId) return false;
    }

    if (filter.agentId !== undefined) {
      const agentId = event.entityType === "agent"
        ? event.entityId
        : String(payload?.agentId ?? "");
      if (agentId !== filter.agentId) return false;
    }

    return true;
  }

  // -----------------------------------------------------------------------
  // Inbound response handling (host → worker, response to our outbound call)
  // -----------------------------------------------------------------------

  function handleHostResponse(response: JsonRpcResponse): void {
    const id = response.id;
    if (id === null || id === undefined) return;

    const pending = pendingRequests.get(id);
    if (!pending) return;

    clearTimeout(pending.timer);
    pendingRequests.delete(id);
    pending.resolve(response);
  }

  // -----------------------------------------------------------------------
  // Incoming line handler
  // -----------------------------------------------------------------------

  function handleLine(line: string): void {
    if (!line.trim()) return;

    let message: unknown;
    try {
      message = parseMessage(line);
    } catch (err) {
      if (err instanceof JsonRpcParseError) {
        // Send parse error response
        sendMessage(
          createErrorResponse(
            null,
            JSONRPC_ERROR_CODES.PARSE_ERROR,
            `Parse error: ${err.message}`,
          ),
        );
      }
      return;
    }

    if (isJsonRpcResponse(message)) {
      // This is a response to one of our outbound worker→host calls
      handleHostResponse(message);
    } else if (isJsonRpcRequest(message)) {
      // This is a host→worker RPC call — dispatch it
      handleHostRequest(message as JsonRpcRequest).catch((err) => {
        // Unhandled error in the async handler — send error response
        const errorMessage = err instanceof Error ? err.message : String(err);
        const errorCode = (err as any)?.code ?? PLUGIN_RPC_ERROR_CODES.WORKER_ERROR;
        try {
          sendMessage(
            createErrorResponse(
              (message as JsonRpcRequest).id,
              typeof errorCode === "number" ? errorCode : PLUGIN_RPC_ERROR_CODES.WORKER_ERROR,
              errorMessage,
            ),
          );
        } catch {
          // Cannot send response, stdout may be closed
        }
      });
    } else if (isJsonRpcNotification(message)) {
      // Dispatch host→worker push notifications
      const notif = message as JsonRpcNotification & { method: string; params?: unknown };
      const runNotification = (fn: () => void | Promise<void>) => {
        if (notif.paperclipInvocation) {
          return invocationContextStorage.run(notif.paperclipInvocation, fn);
        }
        return fn();
      };
      if (notif.method === "agents.sessions.event" && notif.params) {
        const event = notif.params as AgentSessionEvent;
        const cb = sessionEventCallbacks.get(event.sessionId);
        if (cb) cb(event);
      } else if (notif.method === "onEvent" && notif.params) {
        // Plugin event bus notifications — dispatch to registered event handlers
        Promise.resolve(runNotification(() => handleOnEvent(notif.params as OnEventParams))).catch((err) => {
          notifyHost("log", {
            level: "error",
            message: `Failed to handle event notification: ${err instanceof Error ? err.message : String(err)}`,
          });
        });
      }
    }
  }

  // -----------------------------------------------------------------------
  // Cleanup
  // -----------------------------------------------------------------------

  function cleanup(): void {
    running = false;

    // Close readline
    if (readline) {
      readline.close();
      readline = null;
    }

    // Reject all pending outbound calls
    for (const [id, pending] of pendingRequests) {
      clearTimeout(pending.timer);
      pending.resolve(
        createErrorResponse(
          id,
          PLUGIN_RPC_ERROR_CODES.WORKER_UNAVAILABLE,
          "Worker RPC host is shutting down",
        ) as JsonRpcResponse,
      );
    }
    pendingRequests.clear();
    sessionEventCallbacks.clear();
  }

  // -----------------------------------------------------------------------
  // Bootstrap: wire up stdin readline
  // -----------------------------------------------------------------------

  let readline: ReadlineInterface | null = createInterface({
    input: stdinStream as NodeJS.ReadableStream,
    crlfDelay: Infinity,
  });

  readline.on("line", handleLine);

  // If stdin closes, we should exit gracefully
  readline.on("close", () => {
    if (running) {
      cleanup();
      if (!options.stdin && !options.stdout) {
        process.exit(0);
      }
    }
  });

  // Handle uncaught errors in the worker process.
  // Only install these when using the real process streams (not in tests
  // where the caller provides custom streams).
  if (!options.stdin && !options.stdout) {
    process.on("uncaughtException", (err) => {
      notifyHost("log", {
        level: "error",
        message: `Uncaught exception: ${err.message}`,
        meta: { stack: err.stack },
      });
      // Give the notification a moment to flush, then exit
      setTimeout(() => process.exit(1), 100);
    });

    process.on("unhandledRejection", (reason) => {
      const message = reason instanceof Error ? reason.message : String(reason);
      const stack = reason instanceof Error ? reason.stack : undefined;
      notifyHost("log", {
        level: "error",
        message: `Unhandled rejection: ${message}`,
        meta: { stack },
      });
    });
  }

  // -----------------------------------------------------------------------
  // Return the handle
  // -----------------------------------------------------------------------

  return {
    get running() {
      return running;
    },

    stop() {
      cleanup();
    },
  };
}