fix(showcase): repair self-contained worker boot + generalize driver registry seams

CR fixes for the fleet worker driverKind→driver registry (PR CopilotKit#5283):

- Fix default (self-contained) worker boot: build the default d6 as a
  registry entry { driver, payloadToInput, aggregateSlugKey } instead of a
  bare driver with no mapper, so startWorkerLoop's construction guard no
  longer throws "Fleet worker has no drivers".
- Thread aggregate-key derivation through DriverRegistryEntry
  (aggregateSlugKey?), defaulting to d6:<slug> so the d6 cell-capture filter
  stays byte-identical while non-d6 kinds can supply their own scheme.
- Add construction-time fail-loud assert that each registry entry's factory
  kind matches its key; raise unknown-driver-kind log to error to match the
  sibling protocol-violation logs.
- Extract shared buildPooledBrowserDrivers consumed by both
  registerAllProbeDrivers and the worker registry; collapse no-op per-kind
  payload-mapper aliases to the single createPayloadToInput.
- Introduce typed DriverKind union (contained to worker/payload-mapper);
  keep contracts.driverKind a string wire boundary.
- Tests: new fleet/orchestrator.test.ts (default-boot equivalence), driver
  construction-guard + custom aggregateSlugKey coverage, and lock-step +
  registry-wiring pins (factory kind == constant).

Author: jpr5

showcase/harness/src/fleet/contracts.ts

@@ -75,9 +75,14 @@ export interface ServiceJobPayload {
   /** The showcase service / integration slug, e.g. `"langgraph-python"`. */
   serviceSlug: string;
   /**
-   * The d6 driver kind that runs the cells. Always `"e2e_d6"` for the
-   * per-service d6 unit, but carried explicitly so the same job shape can
-   * later host other per-service drivers without a payload migration.
+   * The driver kind that runs the cells. The producer currently stamps
+   * `"e2e_d6"` (the per-service d6 unit), but the WORKER now routes by this
+   * field through its `DriverRegistry` — so any registered kind (e2e_d6 /
+   * e2e_deep / e2e_demos / e2e_smoke) dispatches to the matching driver, and an
+   * unregistered kind is reported as a `worker-protocol-violation`. Kept a
+   * `string` (not narrowed to the worker's `DriverKind` union) because this is
+   * the WIRE boundary that receives whatever the producer serialized; the
+   * runtime unknown-kind guard is the validation gate.
    */
   driverKind: string;
   /**

showcase/harness/src/fleet/orchestrator.test.ts

@@ -0,0 +1,168 @@
+import { describe, it, expect, vi } from "vitest";
+import type { Browser } from "playwright";
+import { runWorker } from "./orchestrator.js";
+import type { FleetRoleConfig } from "./role-config.js";
+import type {
+  FleetQueueClient,
+  ClaimedJob,
+  JobLease,
+  JobView,
+  ServiceJobPayload,
+  ServiceJobResult,
+} from "./contracts.js";
+import type { Logger } from "../types/index.js";
+import type { LaunchBrowser, CgroupPidsReader } from "../probes/helpers/browser-pool.js";
+
+/**
+ * Fleet WORKER entrypoint (`runWorker`, fleet/orchestrator.ts) — DEFAULT
+ * (self-contained) boot equivalence.
+ *
+ * REGRESSION: the default boot path (neither a `drivers` registry nor a legacy
+ * `driver` injected) USED to set a bare `driver = d6Driver` WITHOUT a
+ * `payloadToInput`, so `startWorkerLoop`'s construction guard threw "Fleet
+ * worker has no drivers" and the worker could never boot self-contained. The fix
+ * builds the default d6 as a REGISTRY entry (`e2e_d6 → { driver, payloadToInput,
+ * aggregateSlugKey }`), so the self-contained boot SUCCEEDS and routes an
+ * `e2e_d6` job through the d6 driver.
+ *
+ * The d6 driver is the REAL pooled one; we feed it an input with NO declared
+ * features so it returns its green "no D5 features declared" aggregate WITHOUT
+ * touching chromium (the BrowserPool's launcher is a no-op fake injected via the
+ * test-only `launchBrowser` seam, so no real browser ever spawns).
+ */
+
+const silentLogger: Logger = {
+  info: () => {},
+  warn: () => {},
+  error: () => {},
+  debug: () => {},
+};
+
+/** A no-op connected Browser — the d6 driver never opens a context in this
+ *  test (zero features), so `init()` just needs a launchable fake. */
+function makeNoopLauncher(): LaunchBrowser {
+  return async () =>
+    ({
+      isConnected: () => true,
+      on: () => {},
+      async close() {},
+      async newContext() {
+        return { async close() {} };
+      },
+    }) as unknown as Browser;
+}
+
+/** Always reports plenty of headroom so the claim gate never blocks. */
+const headroomPidsReader: CgroupPidsReader = () => ({ current: 10, max: 1000 });
+
+function makeJobView(overrides: Partial<JobView> = {}): JobView {
+  return {
+    id: "job-1",
+    probe_key: "d6:langgraph-python",
+    status: "claimed",
+    claimed_by: "worker-test",
+    lease_expires_at: "2026-06-04T00:05:00.000Z",
+    version: 1,
+    ...overrides,
+  };
+}
+
+function makePayload(
+  overrides: Partial<ServiceJobPayload> = {},
+): ServiceJobPayload {
+  return {
+    probeKey: "d6:langgraph-python",
+    serviceSlug: "langgraph-python",
+    driverKind: "e2e_d6",
+    // A runnable d6 input with NO declared features → the d6 driver returns its
+    // green "no D5 features declared" aggregate without acquiring a browser.
+    driverInputs: {
+      key: "e2e_d6:langgraph-python",
+      backendUrl: "https://lg.example.com",
+    },
+    meta: {
+      runId: "run-42",
+      triggered: false,
+      enqueuedAt: "2026-06-04T00:00:00.000Z",
+    },
+    ...overrides,
+  };
+}
+
+function makeLease(payload?: Partial<ServiceJobPayload>): JobLease {
+  return {
+    job: makeJobView(),
+    payload: makePayload(payload),
+    leaseExpiresAt: "2026-06-04T00:05:00.000Z",
+  };
+}
+
+interface RecordingQueue extends FleetQueueClient {
+  reports: ServiceJobResult[];
+}
+
+/** A queue fake that hands out a fixed sequence of claims, then idles. */
+function makeQueue(claims: ClaimedJob[]): RecordingQueue {
+  const reports: ServiceJobResult[] = [];
+  let i = 0;
+  return {
+    reports,
+    async enqueue() {
+      throw new Error("enqueue not used by worker");
+    },
+    async claimNext(): Promise<ClaimedJob> {
+      const next = claims[i] ?? { claimed: false };
+      if (i < claims.length) i++;
+      return next;
+    },
+    async renewLease(): Promise<JobLease | null> {
+      return makeLease();
+    },
+    async report({ result }): Promise<void> {
+      reports.push(result);
+    },
+    async sweepExpired() {
+      return { reclaimed: 0, commErrors: [] };
+    },
+  };
+}
+
+const config: FleetRoleConfig = { role: "worker", poolCount: 1 };
+
+describe("runWorker default (self-contained) boot", () => {
+  it("boots WITHOUT an injected registry or driver and routes an e2e_d6 job through the d6 driver", async () => {
+    const queue = makeQueue([{ claimed: true, lease: makeLease() }]);
+
+    // No `drivers`, no `driver`, no `budgetSource` → the default boot path
+    // constructs its own pool + the default d6 REGISTRY entry. Pre-fix this
+    // threw "Fleet worker has no drivers"; post-fix it boots and routes.
+    const worker = await runWorker(config, {
+      queue,
+      workerId: "worker-test",
+      logger: silentLogger,
+      env: {},
+      skipHealthServer: true,
+      leaseSeconds: 2000,
+      heartbeatMs: 1_000_000,
+      pollIntervalMs: 1,
+      // Test-only seams so the default-boot BrowserPool never spawns chromium.
+      launchBrowser: makeNoopLauncher(),
+      cgroupPidsReader: headroomPidsReader,
+    });
+
+    try {
+      await vi.waitFor(() => expect(queue.reports).toHaveLength(1), {
+        timeout: 5000,
+      });
+    } finally {
+      await worker.stop();
+    }
+
+    const result = queue.reports[0]!;
+    // The e2e_d6 job was routed to the REAL d6 driver (not a protocol
+    // violation): the d6 driver's green "no D5 features declared" aggregate.
+    expect(result.commError).toBeUndefined();
+    expect(result.aggregateState).toBe("green");
+    expect(result.aggregateKey).toBe("e2e_d6:langgraph-python");
+  });
+});

showcase/harness/src/fleet/orchestrator.ts

@@ -29,6 +29,10 @@ import { serve } from "@hono/node-server";
 import { createEventBus } from "../events/event-bus.js";
 import { logger as defaultLogger } from "../logger.js";
 import { BrowserPool } from "../probes/helpers/browser-pool.js";
+import type {
+  LaunchBrowser,
+  CgroupPidsReader,
+} from "../probes/helpers/browser-pool.js";
 import {
   createE2eFullDriver,
   createPooledE2eFullLauncher,
@@ -44,6 +48,10 @@ import {
   type DriverRegistry,
   type WorkerLoopHandle,
 } from "./worker/worker-loop.js";
+import {
+  createD6PayloadToInput,
+  E2E_D6_DRIVER_KIND,
+} from "./worker/payload-mapper.js";
 
 /** The worker boot handle — symmetric with the control-plane `boot()` shape. */
 export interface WorkerHandle {
@@ -132,6 +140,20 @@ export interface RunWorkerOptions {
   budgetSource?: {
     budget(): import("../probes/helpers/browser-pool.js").BrowserPoolBudget;
   };
+  /**
+   * Test-only seam: injected chromium launcher forwarded to the `BrowserPool`
+   * the DEFAULT (self-contained) boot path constructs, so the default-boot
+   * equivalence test can exercise that path WITHOUT spawning real chromium.
+   * Production omits it → the pool uses the real launcher. Ignored when a
+   * `budgetSource` + run path are injected (no pool is constructed).
+   */
+  launchBrowser?: LaunchBrowser;
+  /**
+   * Test-only seam: injected cgroup PID reader forwarded to the default-boot
+   * `BrowserPool` (powers `budget()`), so the default-boot test reports
+   * headroom deterministically. Production omits it → the real reader is used.
+   */
+  cgroupPidsReader?: CgroupPidsReader;
 }
 
 /**
@@ -185,8 +207,12 @@ export async function runWorker(
   //      injected `budgetSource`, the worker runs entirely on the caller's wiring
   //      and this entrypoint constructs NO pool of its own.
   //   2. A legacy single `driver` + `budgetSource` (back-compat test path).
-  //   3. Neither → construct the worker's own pool + default pooled d6 driver
-  //      (the original self-contained boot).
+  //   3. Neither → construct the worker's own pool and a DEFAULT REGISTRY holding
+  //      the single pooled d6 driver under its `e2e_d6` kind (the self-contained
+  //      boot). Building it as a registry entry — not a bare `driver` — means the
+  //      self-contained path also routes by driverKind AND carries the d6
+  //      payload mapper, so `startWorkerLoop`'s construction guard is satisfied
+  //      (equivalence with the pre-registry single-d6 worker).
   // The pool is the self-bounded context budget that gates claiming and keeps the
   // worker under its pids ceiling.
   let pool: BrowserPool | undefined;
@@ -196,20 +222,33 @@ export async function runWorker(
 
   // We have all the wiring we need iff a budget source exists AND at least one
   // run path (a registry or a single driver) is supplied. Otherwise build the
-  // default pool + d6 driver so the worker is self-contained.
+  // default pool + d6 registry so the worker is self-contained.
   const haveRunPath = drivers !== undefined || driver !== undefined;
   if (!budgetSource || !haveRunPath) {
-    pool = new BrowserPool({ logger });
+    pool = new BrowserPool({
+      logger,
+      launchBrowser: opts.launchBrowser,
+      cgroupPidsReader: opts.cgroupPidsReader,
+    });
     await pool.init();
     budgetSource = budgetSource ?? pool;
     if (!haveRunPath) {
-      // Default to the single pooled d6 driver registered under its kind, so the
-      // self-contained boot path also routes by driverKind (equivalence with the
-      // pre-registry single-d6 worker).
-      const d6Driver = createE2eFullDriver({
-        launcher: createPooledE2eFullLauncher(pool, logger),
-      });
-      driver = d6Driver;
+      // Default to the single pooled d6 driver registered under its kind, paired
+      // with the d6 payload→input mapper, so the self-contained boot routes by
+      // driverKind through the registry (equivalence with the pre-registry
+      // single-d6 worker) and the loop's construction guard is satisfied.
+      drivers = new Map([
+        [
+          E2E_D6_DRIVER_KIND,
+          {
+            driver: createE2eFullDriver({
+              launcher: createPooledE2eFullLauncher(pool, logger),
+            }),
+            payloadToInput: createD6PayloadToInput(),
+            aggregateSlugKey: (serviceSlug: string) => `d6:${serviceSlug}`,
+          },
+        ],
+      ]);
     }
   }
 

showcase/harness/src/fleet/worker/payload-mapper.test.ts

@@ -1,9 +1,6 @@
 import { describe, expect, it } from "vitest";
 import {
   createD6PayloadToInput,
-  createDeepPayloadToInput,
-  createDemosPayloadToInput,
-  createSmokePayloadToInput,
   createPayloadToInput,
   E2E_D6_DRIVER_KIND,
   E2E_DEEP_DRIVER_KIND,
@@ -82,38 +79,32 @@ describe("driver-kind constants", () => {
   });
 });
 
-describe("per-kind payload mappers", () => {
+describe("shared payload mapper", () => {
   // The four browser driver families share the SAME re-hydration logic (each
   // serializes a `{ key, backendUrl, … }` object and validates via its own zod
-  // schema), so the per-kind mappers all re-hydrate the input the same way.
+  // schema), so every registry entry wires the single `createPayloadToInput`.
+  // `createD6PayloadToInput` is retained as a back-compat alias of it.
   const driverInputs = {
     key: "k:langgraph-python",
     backendUrl: "https://lg.example.com",
   };
 
-  it.each([
-    ["deep", createDeepPayloadToInput],
-    ["demos", createDemosPayloadToInput],
-    ["smoke", createSmokePayloadToInput],
-    ["base", createPayloadToInput],
-  ])("%s mapper re-hydrates the serialized input", (_name, factory) => {
-    const map = factory();
+  it("re-hydrates the serialized input", () => {
+    const map = createPayloadToInput();
     const input = map(payload({ driverInputs })) as Record<string, unknown>;
     expect(input.key).toBe("k:langgraph-python");
     expect(input.backendUrl).toBe("https://lg.example.com");
   });
 
-  it("each per-kind mapper defaults a missing key to the payload probeKey", () => {
-    for (const factory of [
-      createDeepPayloadToInput,
-      createDemosPayloadToInput,
-      createSmokePayloadToInput,
-    ]) {
-      const map = factory();
-      const input = map(
-        payload({ driverInputs: { backendUrl: "https://lg.example.com" } }),
-      ) as Record<string, unknown>;
-      expect(input.key).toBe("d6:langgraph-python");
-    }
+  it("defaults a missing key to the payload probeKey", () => {
+    const map = createPayloadToInput();
+    const input = map(
+      payload({ driverInputs: { backendUrl: "https://lg.example.com" } }),
+    ) as Record<string, unknown>;
+    expect(input.key).toBe("d6:langgraph-python");
+  });
+
+  it("createD6PayloadToInput is the same factory (back-compat alias)", () => {
+    expect(createD6PayloadToInput).toBe(createPayloadToInput);
   });
 });