feat(harness/fleet): generalize enumerator + control-plane for N producer schedules (CopilotKit#5285)

## Summary

Producer-side foundation for fleet framework item 3b. Two
**behavior-preserving** generalizations that make the seam capable of
multiple browser families and multiple producer cadences, while keeping
the d6 case **byte-identical**. No wiring is flipped on yet (see Out of
Scope).

### 1. Parameterized service enumerator
`showcase/harness/src/fleet/control-plane/catalog-enumerator.ts`
- New generic `createServiceEnumerator(params)`
(catalog-enumerator.ts:215) takes the service-set `filter`, the
`driverKind`, and a `probeKeyPrefix` (string prefix → `<prefix>:<slug>`,
or a builder fn).
- `createD6ServiceEnumerator` (catalog-enumerator.ts:280) is
re-expressed as a thin call passing the d6 params: `D6_DRIVER_KIND`
(`e2e_d6`), prefix `"d6"` (→ `d6:<slug>`), and `D6_DISCOVERY_FILTER`. d6
output is unchanged — same services, same filter, same kind, same keys.

### 2. Control-plane accepts an array of producer schedules
`showcase/harness/src/fleet/control-plane/control-plane.ts`
- New `ProducerSchedule` type (`{ scheduleId, cron, producer }`) + a
`schedules?` dep on `ControlPlaneDeps`.
- `createControlPlane` normalizes to an array (control-plane.ts:~232);
omitting `schedules` degenerates to the single d6 schedule on
`FLEET_PRODUCER_SCHEDULE_ID` (`fleet-job-producer`) @ `40 * * * *` —
current behavior preserved exactly.
- `start()` / `stop()` iterate the array, registering/unregistering each
scheduler entry and starting/stopping each producer.

## Out of scope (deferred — gated on other in-flight PRs)
- **No `runControlPlane` wiring** to actually PASS multiple schedules —
that edit conflicts with in-flight **CopilotKit#5284** (which edits
`runControlPlane`) and is deferred. This PR only makes
`control-plane.ts` *capable* of N schedules + generalizes the enumerator
seam; the wiring lands later.
- No `e2e_smoke` / `e2e_demos` / `e2e_deep` enumerators or producers
(Phase 2).
- No changes to `worker-loop.ts` / `payload-mapper.ts` /
`probe-loader.ts` (other PRs own those).
- No driverKind constant / contract changes.

## Test plan
- [x] Red→green TDD: 3 new enumerator tests (generic
kind/keys/filter/fn-prefix) + 2 new control-plane tests (N entries
registered with distinct crons; stop tears all down) failed before impl,
pass after.
- [x] Equivalence: all pre-existing d6-enumerator + single-schedule
control-plane tests pass unchanged.
- [x] Full harness suite green: **2078 passed** (119 files).
- [x] `tsc -p tsconfig.build.json` clean (exit 0).
- [x] Only the 4 intended files changed; no lockfile drift.

Do not merge — producer-side foundation only; wiring follows after CopilotKit#5284
lands.

Author: jpr5

showcase/harness/src/fleet/control-plane/catalog-enumerator.test.ts

@@ -1,6 +1,7 @@
 import { describe, it, expect } from "vitest";
 import {
   createD6ServiceEnumerator,
+  createServiceEnumerator,
   D6_DRIVER_KIND,
   D6_DISCOVERY_FILTER,
 } from "./catalog-enumerator.js";
@@ -207,3 +208,129 @@ describe("createD6ServiceEnumerator", () => {
     expect(specs.length).toBe(3);
   });
 });
+
+describe("createServiceEnumerator (generic seam)", () => {
+  it("stamps the passed driverKind and probeKey prefix (non-d6 params)", async () => {
+    const source = fakeSource([
+      svc({ name: "showcase-langgraph-python" }),
+      svc({ name: "showcase-crewai", publicUrl: "http://crewai:10000" }),
+    ]);
+    const enumerate = createServiceEnumerator({
+      source,
+      env: {},
+      fetchImpl: globalThis.fetch,
+      logger: SILENT_LOGGER,
+      driverKind: "e2e_smoke",
+      probeKeyPrefix: "smoke",
+      filter: { namePrefix: "showcase-" },
+    });
+
+    const specs = await enumerate(CTX);
+
+    expect(specs).toHaveLength(2);
+    const lg = specs.find((s) => s.serviceSlug === "langgraph-python");
+    expect(lg).toBeDefined();
+    expect(lg?.driverKind).toBe("e2e_smoke");
+    expect(lg?.probeKey).toBe("smoke:langgraph-python");
+    expect(lg?.driverInputs?.key).toBe("smoke:langgraph-python");
+    expect(lg?.driverInputs?.backendUrl).toBe("http://langgraph-python:10000");
+
+    const cr = specs.find((s) => s.serviceSlug === "crewai");
+    expect(cr?.probeKey).toBe("smoke:crewai");
+    expect(cr?.driverKind).toBe("e2e_smoke");
+  });
+
+  it("passes the param-supplied filter (namePrefix + nameExcludes) to the source", async () => {
+    const source = fakeSource([svc()]);
+    const filter = {
+      namePrefix: "showcase-",
+      nameExcludes: ["showcase-harness", "showcase-shell"],
+    } as const;
+    const enumerate = createServiceEnumerator({
+      source,
+      env: {},
+      fetchImpl: globalThis.fetch,
+      logger: SILENT_LOGGER,
+      driverKind: "e2e_smoke",
+      probeKeyPrefix: "smoke",
+      filter,
+    });
+
+    await enumerate(CTX);
+
+    expect(source.configs).toHaveLength(1);
+    const cfg = source.configs[0] as {
+      namePrefix?: string;
+      nameExcludes?: string[];
+    };
+    expect(cfg.namePrefix).toBe("showcase-");
+    expect(cfg.nameExcludes).toEqual([...filter.nameExcludes]);
+  });
+
+  it("accepts a function probeKey prefix builder", async () => {
+    const source = fakeSource([svc({ name: "showcase-langgraph-python" })]);
+    const enumerate = createServiceEnumerator({
+      source,
+      env: {},
+      fetchImpl: globalThis.fetch,
+      logger: SILENT_LOGGER,
+      driverKind: "e2e_deep",
+      probeKeyPrefix: (slug) => `deep-${slug}`,
+      filter: { namePrefix: "showcase-" },
+    });
+
+    const specs = await enumerate(CTX);
+    const lg = specs.find((s) => s.serviceSlug === "langgraph-python");
+    expect(lg?.probeKey).toBe("deep-langgraph-python");
+    expect(lg?.driverInputs?.key).toBe("deep-langgraph-python");
+  });
+
+  // A5: a filter without a namePrefix would make the discovery source enumerate
+  // ALL services (a documented incident class). Fail loud at construction.
+  it("throws when the filter has no namePrefix (fail loud, never enumerate all)", () => {
+    const source = fakeSource([svc()]);
+    expect(() =>
+      createServiceEnumerator({
+        source,
+        env: {},
+        fetchImpl: globalThis.fetch,
+        logger: SILENT_LOGGER,
+        driverKind: "e2e_smoke",
+        probeKeyPrefix: "smoke",
+        filter: { nameExcludes: ["showcase-harness"] },
+      }),
+    ).toThrow(/namePrefix/);
+  });
+
+  it("throws when the filter namePrefix is an empty string", () => {
+    const source = fakeSource([svc()]);
+    expect(() =>
+      createServiceEnumerator({
+        source,
+        env: {},
+        fetchImpl: globalThis.fetch,
+        logger: SILENT_LOGGER,
+        driverKind: "e2e_smoke",
+        probeKeyPrefix: "smoke",
+        filter: { namePrefix: "" },
+      }),
+    ).toThrow(/namePrefix/);
+  });
+
+  // A6: a function-form probeKeyPrefix returning "" yields an empty
+  // probeKey/driverInputs.key (a bad join key) — fail loud naming the slug.
+  it("throws when a function probeKeyPrefix yields an empty key for a service", async () => {
+    const source = fakeSource([svc({ name: "showcase-langgraph-python" })]);
+    const enumerate = createServiceEnumerator({
+      source,
+      env: {},
+      fetchImpl: globalThis.fetch,
+      logger: SILENT_LOGGER,
+      driverKind: "e2e_deep",
+      probeKeyPrefix: () => "",
+      filter: { namePrefix: "showcase-" },
+    });
+
+    await expect(enumerate(CTX)).rejects.toThrow(/langgraph-python/);
+  });
+});

showcase/harness/src/fleet/control-plane/catalog-enumerator.ts

@@ -93,6 +93,16 @@ export const D6_DISCOVERY_FILTER = {
   ],
 } as const;
 
+/**
+ * A discovery service-set filter — the `namePrefix` / `nameExcludes` block the
+ * discovery source narrows on. Shared by the d6 wrapper deps and the generic
+ * enumerator params so the shape stays in one place.
+ */
+export interface ServiceSetFilter {
+  namePrefix?: string;
+  nameExcludes?: readonly string[];
+}
+
 export interface D6ServiceEnumeratorDeps {
   /** The discovery source to enumerate (production: `railwayServicesSource`). */
   source: DiscoverySource<RailwayServiceInfo>;
@@ -105,7 +115,48 @@ export interface D6ServiceEnumeratorDeps {
    * Service-set filter. Defaults to `D6_DISCOVERY_FILTER`. Exposed so a test
    * (or a future operator config) can narrow the set without editing the SSOT.
    */
-  filter?: { namePrefix?: string; nameExcludes?: readonly string[] };
+  filter?: ServiceSetFilter;
+}
+
+/**
+ * Build the dashboard `probeKey` for a given service slug. A bare string is
+ * treated as a prefix (`<prefix>:<slug>` — the d6 `d6:<slug>` convention); a
+ * function gives a family full control over the key shape. The same value is
+ * stamped onto `driverInputs.key` so the driver emits the exact dashboard keys.
+ */
+export type ProbeKeyPrefix = string | ((slug: string) => string);
+
+export interface ServiceEnumeratorParams {
+  /** The discovery source to enumerate (production: `railwayServicesSource`). */
+  source: DiscoverySource<RailwayServiceInfo>;
+  /** Frozen env snapshot threaded to the discovery context. */
+  env: Readonly<Record<string, string | undefined>>;
+  /** Fetch impl threaded to the discovery context (tests stub network). */
+  fetchImpl: typeof fetch;
+  logger: Logger;
+  /** The driver kind every enumerated spec runs under (e.g. `e2e_d6`). */
+  driverKind: string;
+  /**
+   * Builds the dashboard `probeKey` (and `driverInputs.key`) per service. A
+   * string is used as a `<prefix>:<slug>` prefix; a function gives full control.
+   */
+  probeKeyPrefix: ProbeKeyPrefix;
+  /**
+   * Service-set filter selecting which discovered services this family runs.
+   * Carries the discovery source's `namePrefix` / `nameExcludes` block.
+   */
+  filter: ServiceSetFilter;
+}
+
+/**
+ * Resolve a `ProbeKeyPrefix` into the concrete dashboard key for a slug. The
+ * string arm OWNS the `:` separator (`<prefix>:<slug>`), so a string-form caller
+ * passes the bare prefix WITHOUT a trailing `:` (e.g. `"d6"`, not `"d6:"`). The
+ * function arm gives a family full control over the key shape (and is
+ * responsible for its own separators).
+ */
+function buildProbeKey(prefix: ProbeKeyPrefix, slug: string): string {
+  return typeof prefix === "function" ? prefix(slug) : `${prefix}:${slug}`;
 }
 
 /**
@@ -148,17 +199,36 @@ function toDriverInputs(
 }
 
 /**
- * Build the real d6 `ServiceEnumerator`. Each call enumerates the discovery
- * source under the d6 filter and maps every service to one `ServiceJobSpec`.
- * The `EnumerateContext.filter` (operator slug scoping on a triggered run) is
- * applied AFTER discovery so an operator can scope a manual run to a subset of
- * services without re-querying — mirroring the in-process invoker's slug filter.
+ * Build a generic per-service `ServiceEnumerator` parameterized by the
+ * service-set filter, the `driverKind`, and the `probeKey` prefix builder. Each
+ * call enumerates the discovery source under the filter and maps every service
+ * to one `ServiceJobSpec`. The `EnumerateContext.filter` (operator slug scoping
+ * on a triggered run) is applied AFTER discovery so an operator can scope a
+ * manual run to a subset of services without re-querying — mirroring the
+ * in-process invoker's slug filter.
+ *
+ * `createD6ServiceEnumerator` is the d6 specialization of this seam; other
+ * browser families re-express their enumerators the same way, each passing its
+ * own filter / kind / prefix.
  */
-export function createD6ServiceEnumerator(
-  deps: D6ServiceEnumeratorDeps,
+export function createServiceEnumerator(
+  params: ServiceEnumeratorParams,
 ): ServiceEnumerator {
-  const { source, env, fetchImpl, logger } = deps;
-  const filter = deps.filter ?? D6_DISCOVERY_FILTER;
+  const { source, env, fetchImpl, logger, driverKind, probeKeyPrefix, filter } =
+    params;
+
+  // Fail loud on a filter with no `namePrefix`: the discovery source treats an
+  // absent/empty prefix as "match everything", which would enumerate ALL
+  // services (the railway-services source documents this incident class). The
+  // d6 wrapper always passes a concrete filter, so it is unaffected. A future
+  // family that legitimately needs "match all" can revisit this; default to
+  // fail-loud now.
+  if (typeof filter.namePrefix !== "string" || filter.namePrefix === "") {
+    throw new Error(
+      `createServiceEnumerator: filter.namePrefix must be a non-empty string ` +
+        `(driverKind ${driverKind}); an absent/empty prefix would enumerate ALL services.`,
+    );
+  }
 
   return async (ctx: EnumerateContext): Promise<ServiceJobSpec[]> => {
     const discoveryCtx: DiscoveryContext = { fetchImpl, env, logger };
@@ -178,11 +248,20 @@ export function createD6ServiceEnumerator(
     for (const svc of services) {
       const slug = deriveSlug(svc.name);
       if (slugSet && !slugSet.has(slug) && !slugSet.has(svc.name)) continue;
-      const probeKey = `d6:${slug}`;
+      const probeKey = buildProbeKey(probeKeyPrefix, slug);
+      // A function-form prefix returning "" yields an empty probeKey (and
+      // `driverInputs.key`) — a bad dashboard/claim join key. Fail loud naming
+      // the offending slug rather than emitting a spec keyed "".
+      if (probeKey === "") {
+        throw new Error(
+          `createServiceEnumerator: probeKeyPrefix produced an empty probeKey ` +
+            `for service "${svc.name}" (slug "${slug}", driverKind ${driverKind}).`,
+        );
+      }
       const spec: ServiceJobSpec = {
         probeKey,
         serviceSlug: slug,
-        driverKind: D6_DRIVER_KIND,
+        driverKind,
         driverInputs: toDriverInputs(svc, slug, probeKey),
       };
       // Forward the operator feature-type scoping to the worker as a cell
@@ -197,9 +276,36 @@ export function createD6ServiceEnumerator(
     logger.info("fleet.control-plane.catalog-enumerated", {
       runId: ctx.runId,
       triggered: ctx.triggered,
+      driverKind,
       discovered: services.length,
       enqueueable: specs.length,
     });
     return specs;
   };
 }
+
+/**
+ * Build the real d6 `ServiceEnumerator` — a thin specialization of
+ * `createServiceEnumerator` with the d6 filter, the `e2e_d6` driver kind, and
+ * the `d6:<slug>` probeKey prefix. The produced SPECS are identical to the prior
+ * hardwired implementation (same services, same filter, same kind, same keys);
+ * the only observable change is the diagnostic log line — the generic
+ * enumerator's `fleet.control-plane.catalog-enumerated` log additionally carries
+ * a `driverKind` field (the specs themselves are unchanged).
+ */
+export function createD6ServiceEnumerator(
+  deps: D6ServiceEnumeratorDeps,
+): ServiceEnumerator {
+  const { source, env, fetchImpl, logger } = deps;
+  const filter = deps.filter ?? D6_DISCOVERY_FILTER;
+
+  return createServiceEnumerator({
+    source,
+    env,
+    fetchImpl,
+    logger,
+    driverKind: D6_DRIVER_KIND,
+    probeKeyPrefix: "d6",
+    filter,
+  });
+}