Use Effect fn for base processor helpers

ts/EFFECT_NATIVE_REWRITE_AUDIT.md

@@ -2164,6 +2164,36 @@ Notes:
   - `cd ts && bun run lint`
   - `git diff --check`
 
+### 2026-06-04: Base Processor Effect.fn Helper Slice
+
+- Status: migrated and package-verified.
+- Completed:
+  - `ts/packages/base/src/processor/flow.ts` now exposes `startEffect` and
+    `stopEffect` as Effect values with spans instead of lazy zero-argument
+    Effect methods.
+  - `Flow.runInCompatibilityScopeEffect` is now a named `Effect.fn` helper
+    instead of a method returning nested `Effect.gen`.
+  - The public `Flow` shape is explicit so Effect error and requirement
+    channels do not widen through recursive object inference.
+  - Producer, request/response, and parameter specs are generic over the
+    receiving flow requirements, so reusable specs remain mountable in flows
+    with additional service dependencies without type assertions.
+  - `ts/packages/base/src/processor/flow-processor.ts` now defines
+    `startFlow`, `configureFlows`, `processNextConfigPush`, the safe consume
+    wrapper, and the Promise `start` compatibility helper with named
+    `Effect.fn` functions.
+  - The focused scan for callable `return Effect.gen(...)` helpers in
+    `flow.ts` is clean. Remaining `flow-processor.ts` matches are one-shot
+    program values / exported program factories and should be scoped
+    separately from reusable helper normalization.
+- Verification:
+  - `cd ts/packages/base && bunx --bun vitest run src/__tests__/flow-spec-runtime.test.ts src/__tests__/flow-processor-runtime.test.ts`
+  - `cd ts && bun run check:tsgo`
+  - `cd ts && bun run build`
+  - `cd ts && bun run test`
+  - `cd ts && bun run lint`
+  - `git diff --check`
+
 ## Subagent Findings To Preserve
 
 - MCP/workbench:
@@ -2198,6 +2228,10 @@ Notes:
   - Flow service startup facades now consistently use `ManagedRuntime`, and
     local scripts should delegate to `runMain()` instead of adding local
     `.catch(console.error/process.exit)` wrappers.
+  - Base `Flow.startEffect` / `stopEffect` are now Effect values, not
+    zero-argument lazy Effect methods. New runtime APIs should expose Effect
+    values where no arguments are needed, and reserve `Effect.fn` for reusable
+    helpers that take arguments.
   - Persistence IO should move toward `FileSystem` or `KeyValueStore` where
     the installed beta has the needed provider surface.
 - Base messaging/processors:
@@ -2233,6 +2267,10 @@ Notes:
   - Base metrics are now Effect-native and Prometheus-formatted through
     `PrometheusMetrics.format`; do not reopen `prom-client` unless a future
     scrape requirement cannot be represented by Effect metrics.
+  - The scratch note's `prom-client` claim is stale: `ts/packages` has no
+    `prom-client` matches, and `metrics/prometheus.ts` already uses
+    `effect/Metric` plus `PrometheusMetrics.format`. Remaining observability
+    work is OTLP/span coverage and wiring existing consumer metrics helpers.
   - Numeric public timeout fields such as `timeoutMs` remain compatibility
     surfaces. Internal runtime config with `Config.number(...Ms)` is still a
     valid `Config.duration` / `Duration` cleanup target.
@@ -2241,6 +2279,9 @@ Notes:
     Socket errors/JSON parsing now use tagged errors and Schema decoding.
     The remaining client `newableFactory` assertions are documented as public
     API compatibility boundaries for this loop.
+  - `EffectRpcClient` remains Promise-shaped at the public facade. A future
+    client API slice can add an Effect/Stream-native surface beside the
+    Promise facade, then migrate callback users incrementally.
   - Gateway/client `DispatchError` contracts now use `S.TaggedErrorClass`; do
     not reopen `S.ErrorClass` unless a new production match appears.
   - Gateway `DispatchStream` now uses Effect-native dispatcher streaming
@@ -2288,6 +2329,9 @@ Notes:
     complete, and the Effect AI stream-part adapter now uses `effect/Match`.
     The remaining provider-layer item is parity-backed Effect AI adapter work,
     not a direct SDK swap.
+  - AI service modernization remains partially valid: Claude already uses
+    `@effect/ai-anthropic`, while OpenAI/Azure/OpenAI-compatible/Mistral/Ollama
+    still need parity-backed Effect AI adapters or provider-specific bridges.
 - Scratch-note follow-ups:
   - `Term` / compact client term serialization is complete for base schema,
     gateway translation, and pure term helper switches. Future work should
@@ -2296,6 +2340,14 @@ Notes:
   - Messaging runtime `Config.duration` cleanup is complete. Internal runtime
     config uses `Duration.Duration`; public timeout compatibility inputs and
     broker receive/error payload boundaries remain numeric milliseconds.
+  - CLI modernization remains valid, but the live installed target is
+    `effect/unstable/cli` rather than an installed `@effect/cli` package.
+  - Chunking remains a small valid `effect/Chunk` slice: the recursive
+    splitter is still array/mutation based and can expose `Chunk.Chunk<string>`
+    internally while preserving service behavior.
+  - Knowledge core internals are largely Effect-native, but public core service
+    facades still expose Promise methods; migrate tests to Effect-first
+    methods before shrinking those facades.
   - Qdrant graph/doc known-collection caches now use
     `MutableHashSet<string>`. Short-lived local traversal sets remain no-ops.
   - Gateway dispatcher static service registries, streaming membership, and
@@ -2322,8 +2374,10 @@ Notes:
   - Fresh strict signal sweep after the 2026-06-04 helper and collection
     slices found no production normal `Error`, raw `try`/`catch`, native
     `switch`, or Effect-focused type assertions under `ts/packages`.
-  - Remaining real helper-normalization targets from the fresh sweep are base
+  - Remaining real helper-normalization targets from the fresh sweep are one
-    processor flow helpers and one workbench atom helper.
+    workbench atom helper plus separately scoped inline callback/program
+    factories in messaging compatibility facades, gateway/librarian helpers,
+    and CLI command actions.
   - Remaining real long-lived native collection targets include base processor
     registries, Librarian service / collection manager state, prompt template
     cache, and a workbench module cache. Local traversal sets and test fakes

ts/packages/base/src/__tests__/flow-spec-runtime.test.ts

@@ -167,7 +167,7 @@ describe("Effect-native flow specifications", () => {
         provideRuntime(
           backend,
           Effect.gen(function* () {
-            yield* flow.startEffect();
+            yield* flow.startEffect;
             const producer = yield* flow.producerEffect(outputProducerSpec);
             const duplicateSpecError = yield* flow.producerEffect(duplicateOutputProducerSpec).pipe(Effect.flip);
             expect(duplicateSpecError._tag).toBe("FlowResourceNotFoundError");
@@ -212,7 +212,7 @@ describe("Effect-native flow specifications", () => {
         provideRuntime(
           backend,
           Effect.gen(function* () {
-            yield* flow.startEffect();
+            yield* flow.startEffect;
             yield* Effect.yieldNow;
             yield* TestClock.adjust(Duration.millis(5));
           }),
@@ -254,7 +254,7 @@ describe("Effect-native flow specifications", () => {
         provideRuntime(
           backend,
           Effect.gen(function* () {
-            yield* flow.startEffect();
+            yield* flow.startEffect;
             const duplicateSpecError = yield* flow.requestorEffect(duplicateRequestResponseSpec).pipe(Effect.flip);
             expect(duplicateSpecError._tag).toBe("FlowResourceNotFoundError");
             const requestor = flow.requestor(requestResponseSpec);
@@ -293,7 +293,7 @@ describe("Effect-native flow specifications", () => {
         provideRuntime(
           backend,
           Effect.gen(function* () {
-            yield* flow.startEffect();
+            yield* flow.startEffect;
             const producerError = yield* flow.producerEffect("missing-producer").pipe(Effect.flip);
             const parameter = yield* flow.parameterEffect(presentParameter);
             const legacyParameter = yield* flow.parameterEffect("present");

ts/packages/base/src/messaging/runtime.ts

@@ -625,7 +625,7 @@ export const RequestResponseFactoryLive = Layer.effect(
 export const runFlowRuntimeScoped = Effect.fn("FlowRuntime.run")(function* <Requirements = never>(
   flow: Flow<Requirements>,
 ) {
-  yield* flow.startEffect().pipe(
+  yield* flow.startEffect.pipe(
     Effect.mapError((error) => flowRuntimeError(flow.name, "start", error)),
   );
   yield* Effect.addFinalizer(() =>
@@ -673,8 +673,8 @@ export const runEffectRequestResponseScoped = Effect.fn("runEffectRequestRespons
   return yield* makeEffectRequestResponseFromPubSub<TReq, TRes>(pubsub, config, options);
 });
 
-export const runFlowScoped = Effect.fn("runFlowScoped")(function* (
+export const runFlowScoped = Effect.fn("runFlowScoped")(function* <Requirements = never>(
-  flow: Flow,
+  flow: Flow<Requirements>,
 ) {
   yield* runFlowRuntimeScoped(flow);
 });

ts/packages/base/src/processor/flow-processor.ts

@@ -37,7 +37,7 @@ import {
 } from "../messaging/runtime.js";
 import { makePubSubService, PubSub } from "../backend/pubsub.js";
 import { loadMessagingRuntimeConfig } from "../runtime/index.ts";
-import { Duration, Effect, Exit, Layer, ManagedRuntime, Scope } from "effect";
+import { Context, Duration, Effect, Exit, Layer, ManagedRuntime, Scope } from "effect";
 import * as O from "effect/Option";
 import * as S from "effect/Schema";
 
@@ -172,127 +172,110 @@ export function runFlowProcessorDefinitionScoped<
     );
   };
 
-  const startFlowEffect = (
+  const startFlowEffect = Effect.fn("FlowProcessor.startFlow")(function* (
     name: string,
     definition: FlowDefinition,
-  ): Effect.Effect<
+  ) {
-    ActiveFlow,
+    const flowRuntime = yield* FlowRuntime;
-    FlowRuntimeError,
+    const scope = yield* Scope.make();
-    FlowRuntime | ProducerFactory | ConsumerFactory | RequestResponseFactory | FlowRequirements
+    const flow = new Flow<FlowRequirements>(
-  > =>
+      name,
-    Effect.gen(function* () {
+      options.id,
-      const flowRuntime = yield* FlowRuntime;
+      options.pubsub,
-      const scope = yield* Scope.make();
+      definition,
-      const flow = new Flow<FlowRequirements>(
+      options.specifications,
-        name,
+    );
-        options.id,
+    return yield* flowRuntime.run(flow).pipe(
-        options.pubsub,
+      Scope.provide(scope),
-        definition,
+      Effect.as({ scope } satisfies ActiveFlow),
-        options.specifications,
+      Effect.catch((error) =>
-      );
+        Scope.close(scope, Exit.void).pipe(
-      return yield* flowRuntime.run(flow).pipe(
+          Effect.flatMap(() => Effect.fail(error)),
-        Scope.provide(scope),
-        Effect.as({ scope } satisfies ActiveFlow),
-        Effect.catch((error) =>
-          Scope.close(scope, Exit.void).pipe(
-            Effect.flatMap(() => Effect.fail(error)),
-          ),
         ),
-      );
+      ),
-    });
+    );
+  });
 
-  const onConfigureFlowsEffect = (
+  const onConfigureFlowsEffect = Effect.fn("FlowProcessor.configureFlows")(function* (
     config: Record<string, unknown>,
     _version: number,
-  ): Effect.Effect<
+  ) {
-    void,
+    const flowDefs = config.flows;
-    FlowRuntimeError,
+    if (flowDefs === undefined) {
-    FlowRuntime | ProducerFactory | ConsumerFactory | RequestResponseFactory | FlowRequirements
+      yield* Effect.log(`[${options.id}] No flows in config push, skipping`);
-  > =>
+      return;
-    Effect.gen(function* () {
+    }
-      const flowDefs = config.flows;
+    const decodedFlowDefs = decodeFlowDefinitions(flowDefs);
-      if (flowDefs === undefined) {
+    if (O.isNone(decodedFlowDefs)) {
-        yield* Effect.log(`[${options.id}] No flows in config push, skipping`);
+      yield* Effect.logWarning(`[${options.id}] Skipping config push: flows is not an object`);
-        return;
+      return;
-      }
+    }
-      const decodedFlowDefs = decodeFlowDefinitions(flowDefs);
+    const flowDefinitions = decodedFlowDefs.value;
-      if (O.isNone(decodedFlowDefs)) {
-        yield* Effect.logWarning(`[${options.id}] Skipping config push: flows is not an object`);
-        return;
-      }
-      const flowDefinitions = decodedFlowDefs.value;
 
-      const flowsJson = yield* S.encodeUnknownEffect(S.UnknownFromJsonString)(flowDefinitions).pipe(
+    const flowsJson = yield* S.encodeUnknownEffect(S.UnknownFromJsonString)(flowDefinitions).pipe(
-        Effect.catch((error) => Effect.succeed(String(error))),
+      Effect.catch((error) => Effect.succeed(String(error))),
-      );
+    );
-      if (lastFlowsJson.length > 0 && flowsJson === lastFlowsJson && flows.size > 0) {
+    if (lastFlowsJson.length > 0 && flowsJson === lastFlowsJson && flows.size > 0) {
-        yield* Effect.log(`[${options.id}] Flow definitions unchanged, skipping restart`);
+      yield* Effect.log(`[${options.id}] Flow definitions unchanged, skipping restart`);
-        return;
+      return;
-      }
+    }
-      lastFlowsJson = flowsJson;
+    lastFlowsJson = flowsJson;
 
-      for (const [name, activeFlow] of flows) {
+    for (const [name, activeFlow] of flows) {
-        if (!(name in flowDefinitions)) {
+      if (!(name in flowDefinitions)) {
-          yield* Effect.log(`[${options.id}] Stopping removed flow: ${name}`);
+        yield* Effect.log(`[${options.id}] Stopping removed flow: ${name}`);
-          yield* closeFlowEffect(name, activeFlow);
+        yield* closeFlowEffect(name, activeFlow);
-          flows.delete(name);
+        flows.delete(name);
-        }
+      }
+    }
+
+    for (const [name, defn] of Object.entries(flowDefinitions)) {
+      const existing = flows.get(name);
+      if (existing !== undefined) {
+        yield* Effect.log(`[${options.id}] Restarting flow "${name}" with updated config`);
+        yield* closeFlowEffect(name, existing);
+        flows.delete(name);
       }
 
-      for (const [name, defn] of Object.entries(flowDefinitions)) {
+      yield* Effect.log(`[${options.id}] Starting flow "${name}"`);
-        const existing = flows.get(name);
+      const activeFlow = yield* startFlowEffect(name, defn);
-        if (existing !== undefined) {
+      flows.set(name, activeFlow);
-          yield* Effect.log(`[${options.id}] Restarting flow "${name}" with updated config`);
+      yield* Effect.log(`[${options.id}] Flow "${name}" started`);
-          yield* closeFlowEffect(name, existing);
+    }
-          flows.delete(name);
+  });
-        }
 
-        yield* Effect.log(`[${options.id}] Starting flow "${name}"`);
+  const processNextConfigPushEffect = Effect.fn("FlowProcessor.processNextConfigPush")(function* () {
-        const activeFlow = yield* startFlowEffect(name, defn);
+    const consumer = configConsumer;
-        flows.set(name, activeFlow);
+    if (consumer === null) {
-        yield* Effect.log(`[${options.id}] Flow "${name}" started`);
+      yield* Effect.sleep(Duration.millis(1000));
-      }
+      return;
+    }
+
+    const msg = yield* Effect.tryPromise({
+      try: () => consumer.receive(2000),
+      catch: (error) => pubSubError("receive:config-push", error),
     });
+    if (msg === null) {
+      return;
+    }
 
-  const processNextConfigPushEffect = (): Effect.Effect<
+    const push = msg.value();
-    void,
+    yield* Effect.log(`[${options.id}] Received config push version=${push.version}`);
-    never,
-    | FlowRuntime
-    | ProducerFactory
-    | ConsumerFactory
-    | RequestResponseFactory
-    | FlowRequirements
-    | ConfigHandlerRequirements
-  > =>
-    Effect.gen(function* () {
-      const consumer = configConsumer;
-      if (consumer === null) {
-        yield* Effect.sleep(Duration.millis(1000));
-        return;
-      }
 
-      const msg = yield* Effect.tryPromise({
+    yield* onConfigureFlowsEffect(push.config, push.version);
-        try: () => consumer.receive(2000),
-        catch: (error) => pubSubError("receive:config-push", error),
-      });
-      if (msg === null) {
-        return;
-      }
 
-      const push = msg.value();
+    for (const handler of options.configHandlers ?? []) {
-      yield* Effect.log(`[${options.id}] Received config push version=${push.version}`);
+      yield* handler(push.config, push.version);
+    }
 
-      yield* onConfigureFlowsEffect(push.config, push.version);
+    yield* Effect.tryPromise({
+      try: () => consumer.acknowledge(msg),
+      catch: (error) => pubSubError("acknowledge:config-push", error),
+    });
+  });
 
-      for (const handler of options.configHandlers ?? []) {
+  const processNextConfigPushSafelyEffect = Effect.fn("FlowProcessor.processNextConfigPushSafely")(function* () {
-        yield* handler(push.config, push.version);
+    return yield* processNextConfigPushEffect().pipe(
-      }
-
-      yield* Effect.tryPromise({
-        try: () => consumer.acknowledge(msg),
-        catch: (error) => pubSubError("acknowledge:config-push", error),
-      });
-    }).pipe(
       Effect.catch((error) => {
         if (!isRunning()) {
           return Effect.void;
@@ -304,6 +287,7 @@ export function runFlowProcessorDefinitionScoped<
         );
       }),
     );
+  });
 
   return Effect.gen(function* () {
     const pubsub = yield* PubSub;
@@ -325,7 +309,7 @@ export function runFlowProcessorDefinitionScoped<
 
     yield* Effect.whileLoop({
       while: isRunning,
-      body: processNextConfigPushEffect,
+      body: processNextConfigPushSafelyEffect,
       step: () => undefined,
     });
   });
@@ -368,6 +352,24 @@ export function makeFlowProcessor<FlowRequirements = never>(
     return options.provide?.(effect) ?? effect;
   };
 
+  const startProcessorEffect = Effect.fn("FlowProcessor.start")(function* (
+    context: Context.Context<FlowProcessorRuntimeRequirements<FlowRequirements>>,
+  ) {
+    const pubsub = makePubSubService(base.pubsub);
+    const messagingConfig = yield* loadMessagingRuntimeConfig();
+    const start = processor.startEffect.pipe(
+      Effect.provideService(PubSub, pubsub),
+      Effect.provideService(ProducerFactory, ProducerFactory.of(makeProducerFactoryService(pubsub))),
+      Effect.provideService(ConsumerFactory, ConsumerFactory.of(makeConsumerFactoryService(pubsub, messagingConfig))),
+      Effect.provideService(
+        RequestResponseFactory,
+        RequestResponseFactory.of(makeRequestResponseFactoryService(pubsub, messagingConfig)),
+      ),
+      Effect.provideService(FlowRuntime, FlowRuntime.of({ run: runFlowRuntimeScoped })),
+    );
+    return yield* Effect.provide(Effect.scoped(start), context);
+  });
+
   processor = {
     ...base,
     specifications,
@@ -377,25 +379,7 @@ export function makeFlowProcessor<FlowRequirements = never>(
     get startEffect() {
       return makeStartEffect();
     },
-    start: (context) =>
+    start: (context) => compatibilityRuntime.runPromise(startProcessorEffect(context)),
-      compatibilityRuntime.runPromise(Effect.provide(
-        Effect.gen(function* () {
-          const pubsub = makePubSubService(base.pubsub);
-          const messagingConfig = yield* loadMessagingRuntimeConfig();
-          const start = processor.startEffect.pipe(
-            Effect.provideService(PubSub, pubsub),
-            Effect.provideService(ProducerFactory, ProducerFactory.of(makeProducerFactoryService(pubsub))),
-            Effect.provideService(ConsumerFactory, ConsumerFactory.of(makeConsumerFactoryService(pubsub, messagingConfig))),
-            Effect.provideService(
-              RequestResponseFactory,
-              RequestResponseFactory.of(makeRequestResponseFactoryService(pubsub, messagingConfig)),
-            ),
-            Effect.provideService(FlowRuntime, FlowRuntime.of({ run: runFlowRuntimeScoped })),
-          );
-          yield* Effect.scoped(start);
-        }),
-        context,
-      )),
   };
 
   return processor;

ts/packages/base/src/processor/flow.ts

@@ -4,7 +4,7 @@
  * Python reference: trustgraph-base/trustgraph/base/flow.py
  */
 
-import { Context, Effect, Exit, Layer, ManagedRuntime, Scope } from "effect";
+import { Config as EffectConfig, Context, Effect, Exit, Layer, ManagedRuntime, Scope } from "effect";
 import * as O from "effect/Option";
 import * as S from "effect/Schema";
 import type { PubSubBackend } from "../backend/types.js";
@@ -66,13 +66,68 @@ export interface FlowRequestor<TReq, TRes> {
 
 type FlowParameterError = FlowResourceNotFoundError | FlowParameterDecodeError;
 
+export interface Flow<Requirements = never> {
+  readonly name: string;
+  readonly processorId: string;
+  startEffect: Effect.Effect<void, PubSubError, SpecRuntimeRequirements | Requirements>;
+  start: (context: Context.Context<Requirements>) => Promise<void>;
+  stop: () => Promise<void>;
+  stopEffect: Effect.Effect<void>;
+  runInCompatibilityScopeEffect: <A, E>(
+    effect: Effect.Effect<A, E, SpecRuntimeRequirements | Requirements>,
+    runtimePubsub: PubSubBackend,
+    context: Context.Context<Requirements>,
+  ) => Effect.Effect<A, E | EffectConfig.ConfigError>;
+  runInCompatibilityScope: <A, E>(
+    effect: Effect.Effect<A, E, SpecRuntimeRequirements | Requirements>,
+    runtimePubsub: PubSubBackend,
+    context: Context.Context<Requirements>,
+  ) => Promise<A>;
+  clearResources: () => void;
+  registerProducer: <T>(registerName: string, producer: EffectProducer<T>) => void;
+  registerConsumer: (registerName: string, consumer: EffectConsumer) => void;
+  registerRequestor: <TReq, TRes>(
+    registerName: string,
+    rr: EffectRequestResponse<TReq, TRes>,
+  ) => void;
+  setParameter: (parameterName: string, value: unknown) => void;
+  producerEffect: {
+    <T>(producerSpec: ProducerSpec<T>): Effect.Effect<EffectProducer<T>, FlowResourceNotFoundError>;
+    (producerName: string): Effect.Effect<EffectProducer<never>, FlowResourceNotFoundError>;
+  };
+  consumerEffect: (consumerName: string) => Effect.Effect<EffectConsumer, FlowResourceNotFoundError>;
+  requestorEffect: {
+    <TReq, TRes>(
+      requestorSpec: RequestResponseSpec<TReq, TRes>,
+    ): Effect.Effect<EffectRequestResponse<TReq, TRes>, FlowResourceNotFoundError>;
+    (requestorName: string): Effect.Effect<EffectRequestResponse<never, unknown>, FlowResourceNotFoundError>;
+  };
+  parameterEffect: {
+    <T>(parameterSpec: ParameterSpec<T>): Effect.Effect<T, FlowParameterError>;
+    (parameterName: string): Effect.Effect<unknown, FlowResourceNotFoundError>;
+  };
+  producer: {
+    <T>(producerSpec: ProducerSpec<T>): FlowProducer<T>;
+    (producerName: string): FlowProducer<never>;
+  };
+  consumer: (consumerName: string) => FlowConsumer;
+  requestor: {
+    <TReq, TRes>(requestorSpec: RequestResponseSpec<TReq, TRes>): FlowRequestor<TReq, TRes>;
+    (requestorName: string): FlowRequestor<never, unknown>;
+  };
+  parameter: {
+    <T>(parameterSpec: ParameterSpec<T>): T;
+    (parameterName: string): unknown;
+  };
+}
+
 export function makeFlow<Requirements = never>(
   name: string,
   processorId: string,
   pubsub: PubSubBackend,
   definition: FlowDefinition,
   specifications: ReadonlyArray<Spec<Requirements>>,
-) {
+): Flow<Requirements> {
   const producers = new Map<string, EffectProducer<never>>();
   const consumers = new Map<string, EffectConsumer>();
   const requestors = new Map<string, EffectRequestResponse<never, unknown>>();
@@ -275,62 +330,56 @@ export function makeFlow<Requirements = never>(
     return toFlowRequestor(compatibilityRuntime.runSync(requestor.requestorEffect(flow)));
   }
 
-  const flow = {
+  const flow: Flow<Requirements> = {
     name,
     processorId,
-    startEffect(): Effect.Effect<void, PubSubError, SpecRuntimeRequirements | Requirements> {
+    startEffect: Effect.gen(function* () {
-      return Effect.gen(function* () {
+      for (const spec of specifications) {
-        for (const spec of specifications) {
+        yield* spec.addEffect(flow, definition);
-          yield* spec.addEffect(flow, definition);
+      }
-        }
+    }).pipe(Effect.withSpan("Flow.startEffect")),
-      });
-    },
     start(context: Context.Context<Requirements>): Promise<void> {
       return compatibilityRuntime.runPromise(
         Effect.gen(function* () {
           if (compatibilityScope !== null) {
-            yield* flow.stopEffect();
+            yield* flow.stopEffect;
           }
-          yield* flow.runInCompatibilityScopeEffect(flow.startEffect(), pubsub, context);
+          yield* flow.runInCompatibilityScopeEffect(flow.startEffect, pubsub, context);
         }),
       );
     },
     stop(): Promise<void> {
-      return compatibilityRuntime.runPromise(flow.stopEffect());
+      return compatibilityRuntime.runPromise(flow.stopEffect);
     },
-    stopEffect(): Effect.Effect<void> {
+    stopEffect: Effect.gen(function* () {
-      return Effect.gen(function* () {
+      const scope = compatibilityScope;
-        const scope = compatibilityScope;
+      compatibilityScope = null;
-        compatibilityScope = null;
+      if (scope !== null) {
-        if (scope !== null) {
+        yield* Scope.close(scope, Exit.void);
-          yield* Scope.close(scope, Exit.void);
+      }
-        }
+      flow.clearResources();
-        flow.clearResources();
+    }).pipe(Effect.withSpan("Flow.stopEffect")),
-      });
+    runInCompatibilityScopeEffect: Effect.fn("Flow.runInCompatibilityScopeEffect")(function* <A, E>(
-    },
-    runInCompatibilityScopeEffect<A, E>(
       effect: Effect.Effect<A, E, SpecRuntimeRequirements | Requirements>,
       runtimePubsub: PubSubBackend,
       context: Context.Context<Requirements>,
     ) {
-      return Effect.gen(function* () {
+      const scope = yield* ensureCompatibilityScopeEffect();
-        const scope = yield* ensureCompatibilityScopeEffect();
+      const pubsubService = makePubSubService(runtimePubsub);
-        const pubsubService = makePubSubService(runtimePubsub);
+      const messagingConfig = yield* loadMessagingRuntimeConfig();
-        const messagingConfig = yield* loadMessagingRuntimeConfig();
+      return yield* Effect.provide(
-        return yield* Effect.provide(
+        effect.pipe(
-          effect.pipe(
+          Effect.provideService(ProducerFactory, ProducerFactory.of(makeProducerFactoryService(pubsubService))),
-            Effect.provideService(ProducerFactory, ProducerFactory.of(makeProducerFactoryService(pubsubService))),
+          Effect.provideService(ConsumerFactory, ConsumerFactory.of(makeConsumerFactoryService(pubsubService, messagingConfig))),
-            Effect.provideService(ConsumerFactory, ConsumerFactory.of(makeConsumerFactoryService(pubsubService, messagingConfig))),
+          Effect.provideService(
-            Effect.provideService(
+            RequestResponseFactory,
-              RequestResponseFactory,
+            RequestResponseFactory.of(makeRequestResponseFactoryService(pubsubService, messagingConfig)),
-              RequestResponseFactory.of(makeRequestResponseFactoryService(pubsubService, messagingConfig)),
-            ),
-            Scope.provide(scope),
           ),
-          context,
+          Scope.provide(scope),
-        );
+        ),
-      });
+        context,
-    },
+      );
+    }),
     runInCompatibilityScope<A, E>(
       effect: Effect.Effect<A, E, SpecRuntimeRequirements | Requirements>,
       runtimePubsub: PubSubBackend,
@@ -380,8 +429,6 @@ export function makeFlow<Requirements = never>(
   return flow;
 }
 
-export type Flow<Requirements = never> = ReturnType<typeof makeFlow<Requirements>>;
-
 export const Flow = makeFlow as unknown as {
   new <Requirements = never>(
     name: string,

ts/packages/base/src/spec/parameter-spec.ts

@@ -7,16 +7,28 @@
 import { Effect, type Context } from "effect";
 import * as S from "effect/Schema";
 import type { PubSubBackend } from "../backend/types.js";
-import type { Spec } from "./types.js";
+import type { SpecRuntimeRequirements } from "./types.js";
 import type { Flow, FlowDefinition } from "../processor/flow.js";
+import type { PubSubError } from "../errors.js";
 
 declare const ParameterSpecType: unique symbol;
 
 const UnknownParameterSchema: S.Codec<unknown, unknown> = S.Unknown;
 
-export interface ParameterSpec<T = unknown> extends Spec {
+export interface ParameterSpec<T = unknown> {
   readonly [ParameterSpecType]?: (_: T) => T;
+  readonly name: string;
   readonly schema: S.Codec<T, unknown>;
+  readonly addEffect: <Requirements = never>(
+    flow: Flow<Requirements>,
+    definition: FlowDefinition,
+  ) => Effect.Effect<void, PubSubError, SpecRuntimeRequirements | Requirements>;
+  readonly add: <Requirements = never>(
+    flow: Flow<Requirements>,
+    pubsub: PubSubBackend,
+    definition: FlowDefinition,
+    context: Context.Context<Requirements>,
+  ) => Promise<void>;
 }
 
 export function makeParameterSpec(name: string): ParameterSpec<unknown>;
@@ -29,7 +41,7 @@ export function makeParameterSpec<T>(
   schema?: S.Codec<T, unknown>,
 ) {
   const parameterSchema = schema ?? UnknownParameterSchema;
-  const addEffect = (flow: Flow, definition: FlowDefinition) =>
+  const addEffect = <Requirements = never>(flow: Flow<Requirements>, definition: FlowDefinition) =>
     Effect.sync(() => {
       const value = definition.parameters?.[name];
       flow.setParameter(name, value);
@@ -39,11 +51,11 @@ export function makeParameterSpec<T>(
     name,
     schema: parameterSchema,
     addEffect,
-    add: (
+    add: <Requirements = never>(
-      flow: Flow,
+      flow: Flow<Requirements>,
       pubsub: PubSubBackend,
       definition: FlowDefinition,
-      context: Context.Context<never>,
+      context: Context.Context<Requirements>,
     ) =>
       flow.runInCompatibilityScope(addEffect(flow, definition), pubsub, context),
   };

ts/packages/base/src/spec/producer-spec.ts

@@ -4,12 +4,14 @@
  * Python reference: trustgraph-base/trustgraph/base/producer_spec.py
  */
 
-import { Effect } from "effect";
+import { Effect, type Context } from "effect";
-import type { Spec } from "./types.js";
+import type { SpecRuntimeRequirements } from "./types.js";
 import type { Flow, FlowDefinition } from "../processor/flow.js";
+import type { PubSubBackend } from "../backend/types.js";
 import {
   flowResourceNotFoundError,
   type FlowResourceNotFoundError,
+  type PubSubError,
 } from "../errors.js";
 import {
   type EffectProducer,
@@ -18,8 +20,19 @@ import {
 
 declare const ProducerSpecType: unique symbol;
 
-export interface ProducerSpec<T> extends Spec {
+export interface ProducerSpec<T> {
   readonly [ProducerSpecType]?: (_: T) => T;
+  readonly name: string;
+  readonly addEffect: <Requirements = never>(
+    flow: Flow<Requirements>,
+    definition: FlowDefinition,
+  ) => Effect.Effect<void, PubSubError, SpecRuntimeRequirements | Requirements>;
+  readonly add: <Requirements = never>(
+    flow: Flow<Requirements>,
+    pubsub: PubSubBackend,
+    definition: FlowDefinition,
+    context: Context.Context<Requirements>,
+  ) => Promise<void>;
   readonly producerEffect: <Requirements = never>(
     flow: Flow<Requirements>,
   ) => Effect.Effect<EffectProducer<T>, FlowResourceNotFoundError>;
@@ -55,8 +68,8 @@ export function makeProducerSpec<T>(name: string): ProducerSpec<T> {
       : Effect.succeed(producer);
   };
 
-  const addEffect = Effect.fn("ProducerSpec.addEffect")(function* (
+  const addEffect = Effect.fn("ProducerSpec.addEffect")(function* <Requirements = never>(
-    flow: Flow,
+    flow: Flow<Requirements>,
     definition: FlowDefinition,
   ) {
       const topic = definition.topics?.[name] ?? name;

ts/packages/base/src/spec/request-response-spec.ts

@@ -7,12 +7,14 @@
  * Python reference: trustgraph-base/trustgraph/base/prompt_client_spec.py
  */
 
-import { Effect } from "effect";
+import { Effect, type Context } from "effect";
-import type { Spec } from "./types.js";
+import type { SpecRuntimeRequirements } from "./types.js";
 import type { Flow, FlowDefinition } from "../processor/flow.js";
+import type { PubSubBackend } from "../backend/types.js";
 import {
   flowResourceNotFoundError,
   type FlowResourceNotFoundError,
+  type PubSubError,
 } from "../errors.js";
 import {
   type EffectRequestResponse,
@@ -21,11 +23,22 @@ import {
 
 declare const RequestResponseSpecType: unique symbol;
 
-export interface RequestResponseSpec<TReq, TRes> extends Spec {
+export interface RequestResponseSpec<TReq, TRes> {
   readonly [RequestResponseSpecType]?: {
     readonly request: TReq;
     readonly response: TRes;
   };
+  readonly name: string;
+  readonly addEffect: <Requirements = never>(
+    flow: Flow<Requirements>,
+    definition: FlowDefinition,
+  ) => Effect.Effect<void, PubSubError, SpecRuntimeRequirements | Requirements>;
+  readonly add: <Requirements = never>(
+    flow: Flow<Requirements>,
+    pubsub: PubSubBackend,
+    definition: FlowDefinition,
+    context: Context.Context<Requirements>,
+  ) => Promise<void>;
   readonly requestorEffect: <Requirements = never>(
     flow: Flow<Requirements>,
   ) => Effect.Effect<EffectRequestResponse<TReq, TRes>, FlowResourceNotFoundError>;
@@ -65,8 +78,8 @@ export function makeRequestResponseSpec<TReq, TRes>(
       : Effect.succeed(requestor);
   };
 
-  const addEffect = Effect.fn("RequestResponseSpec.addEffect")(function* (
+  const addEffect = Effect.fn("RequestResponseSpec.addEffect")(function* <Requirements = never>(
-    flow: Flow,
+    flow: Flow<Requirements>,
     definition: FlowDefinition,
   ) {
       const requestTopic = definition.topics?.[requestTopicName] ?? requestTopicName;