export const peersLayer = <Self, S extends Spec, EIn = never, RIn = never>(
  tag: HyperlinkTag<Self, S>,
  self: AnyNode,
  options?: {
    /** The fleet (including `self`) — supply it **at the use site** so a shared resource can be defined
     *  node-free and exported; falls back to the tag's baked-in {@link distributed} set when omitted.
     *  An explicit empty array is directory-backed (same as bare {@link distributed}). */
    readonly nodes?: ReadonlyArray<AnyNode>;
    readonly url?: (node: AnyNode) => Effect.Effect<string | undefined, EIn, RIn>;
  },
): Layer.Layer<PeersId<Self> | SelfNodeId<Self>, EIn, RIn> =>
  Layer.merge(
    Layer.effect(
      tag[peersSym],
      Effect.gen(function* () {
        const stamped =
          options?.nodes !== undefined ? options.nodes : tag[nodesSym];

        // D3: stamped empty set → Lookup directory membership (soft if Directory absent).
        // Live rebind on Directory.changes when dial moves / peers appear or leave.
        if (stamped !== undefined && stamped.length === 0) {
          const Lookup = yield* Effect.promise(() => import("./Lookup"));
          const dirOpt = yield* Effect.serviceOption(Lookup.Directory);
          if (Option.isNone(dirOpt)) {
            return {} as Record<string, PeerServiceOf<S>>;
          }
          const directory = dirOpt.value;
          const serviceKey = tag[wireKeySym];
          type DialTarget = {
            readonly key: string;
            readonly kind: ProtocolKind;
            readonly url?: string;
            readonly path?: string;
          };
          const peersRecord = Object.create(null) as Record<
            string,
            PeerServiceOf<S>
          >;
          const peerScopes = new Map<string, Scope.Closeable>();

          const dialTargetOf = (row: {
            readonly nodeKey: string;
            readonly kind: ProtocolKind;
            readonly url?: string;
            readonly path?: string;
          }): DialTarget | undefined => {
            if (row.nodeKey === self.key) return undefined;
            if (row.kind === "IpcSocket" && row.path !== undefined) {
              return { key: row.nodeKey, kind: row.kind, path: row.path };
            }
            if (row.url !== undefined) {
              return { key: row.nodeKey, kind: row.kind, url: row.url };
            }
            return undefined;
          };

          const removePeer = (nodeKey: string): Effect.Effect<void> =>
            Effect.gen(function* () {
              const scope = peerScopes.get(nodeKey);
              if (scope === undefined) return;
              peerScopes.delete(nodeKey);
              delete peersRecord[nodeKey];
              yield* Scope.close(scope, Exit.void);
            });

          const upsertPeer = (row: {
            readonly nodeKey: string;
            readonly kind: ProtocolKind;
            readonly url?: string;
            readonly path?: string;
            readonly serves: ReadonlyArray<string>;
          }): Effect.Effect<void> =>
            Effect.gen(function* () {
              if (!row.serves.includes(serviceKey)) {
                yield* removePeer(row.nodeKey);
                return;
              }
              const target = dialTargetOf(row);
              if (target === undefined) {
                yield* removePeer(row.nodeKey);
                return;
              }
              yield* removePeer(row.nodeKey);
              const scope = yield* Scope.make();
              const client = yield* buildPeerClientAt(tag, target).pipe(
                Scope.provide(scope),
              );
              peerScopes.set(row.nodeKey, scope);
              peersRecord[row.nodeKey] = client;
            });

          const rows = yield* Lookup.nodesServing(tag).pipe(
            Effect.provideService(Lookup.Directory, directory),
          );
          yield* Effect.forEach(rows, upsertPeer, { discard: true });

          yield* directory.changes.pipe(
            Stream.runForEach((event) => {
              if (event._tag === "DirectoryRemoved") {
                return removePeer(event.nodeKey);
              }
              // New peer, dial moved, or serves list may add/remove this service key.
              if (
                event.previous === undefined ||
                event.dialChanged ||
                peersRecord[event.entry.nodeKey] === undefined ||
                !event.entry.serves.includes(serviceKey)
              ) {
                return upsertPeer(event.entry);
              }
              return Effect.void;
            }),
            Effect.forkScoped,
          );

          yield* Effect.addFinalizer(() =>
            Effect.forEach([...peerScopes.keys()], removePeer, {
              discard: true,
            }),
          );

          return peersRecord;
        }

        // Fixed fleet (or undeclared → []); drop self to get the peers.
        const fleet = stamped ?? [];
        const others = fleet.filter((node) => node.key !== self.key);
        const resolveUrl = (
          node: AnyNode,
        ): Effect.Effect<string | undefined, EIn, RIn> =>
          options?.url === undefined
            ? Effect.succeed(node.url)
            : Effect.map(options.url(node), (override) => override ?? node.url);
        const resolved = yield* Effect.forEach(others, (node) =>
          Effect.map(resolveUrl(node), (url) => ({
            key: node.key,
            kind: node.kind,
            url,
            path: node.path,
          })),
        );
        const entries = yield* Effect.forEach(
          // no dialable address → skip (partial mesh); ipc path counts when url absent
          resolved.filter(
            (entry) =>
              entry.url !== undefined ||
              (entry.kind === "IpcSocket" && entry.path !== undefined),
          ),
          (target) =>
            Effect.map(
              buildPeerClientAt(tag, {
                key: target.key,
                kind: target.kind,
                ...(target.url !== undefined ? { url: target.url } : {}),
                ...(target.path !== undefined ? { path: target.path } : {}),
              }),
              (client) => [target.key, client] as const,
            ),
        );
        // Boundary: each peer client is a full `ServiceOf<S>` — a width-supertype of the leaf
        // `PeerServiceOf<S>` the capability exposes — but the mapped types don't reduce under a generic
        // `S`, so TS can't see the overlap; the erasure through `unknown` is the honest boundary.
        return Object.fromEntries(entries) as unknown as Record<string, PeerServiceOf<S>>;
      }),
    ),
    selfNodeLayer(tag, self),
  );