feat(proxy): support multiple local core endpoints, one per line (#13923)

Co-authored-by: Max <maxmad64@gmail.com>

changelog.d/features/13923-plural-local-core-endpoints.md

@@ -0,0 +1 @@
+- **feat(proxy):** support multiple local core endpoints, one per line ([#13923](https://github.com/diegosouzapw/OmniRoute/pull/13923) — thanks @maxmad64bis)

src/app/(dashboard)/dashboard/settings/components/proxy/SubscriptionTab.tsx

@@ -338,7 +338,8 @@ export default function SubscriptionTab() {
 
             <label className="flex flex-col gap-1 text-sm">
               <span className="text-text-muted">{t("proxySubscription.localCoreEndpoint")}</span>
-              <input
+              <textarea
+                rows={3}
                 className="rounded border border-border bg-surface px-2 py-1.5 text-text outline-none focus:border-primary"
                 value={form.localCoreEndpoint}
                 onChange={(e) => setForm({ ...form, localCoreEndpoint: e.target.value })}

src/i18n/messages/en.json

@@ -6852,8 +6852,8 @@
       "globalModeDesc": "All Provider traffic goes through this subscription's proxy pool.",
       "ruleModeDesc": "Only selected Providers' traffic goes through the proxy; the rest connect directly.",
       "localCoreEndpoint": "Local Core SOCKS5/HTTP Endpoint (Optional)",
-      "localCoreEndpointPlaceholder": "socks5://127.0.0.1:1080",
-      "localCoreEndpointDesc": "Only accepts 127.0.0.1 / localhost (SS/VMess/Trojan/VLESS require a local sing-box/clash core).",
+      "localCoreEndpointPlaceholder": "socks5://127.0.0.1:1080\nsocks5://127.0.0.1:1081",
+      "localCoreEndpointDesc": "One per line. Only accepts 127.0.0.1 / localhost (SS/VMess/Trojan/VLESS require a local sing-box/clash core).",
       "routeByProvider": "Route by Provider (multi-select)",
       "loadingProviders": "Loading Provider list…",
       "autoRefreshInterval": "Auto Refresh Interval (minutes)",

src/i18n/messages/vi.json

@@ -6852,8 +6852,8 @@
       "globalModeDesc": "Tất cả lưu lượng truy cập của Nhà cung cấp đều đi qua nhóm proxy của đăng ký này.",
       "ruleModeDesc": "Chỉ lưu lượng truy cập của Nhà cung cấp được chọn mới đi qua proxy; phần còn lại kết nối trực tiếp.",
       "localCoreEndpoint": "Điểm cuối lõi cục bộ SOCKS5/HTTP (Tùy chọn)",
-      "localCoreEndpointPlaceholder": "vớ5://127.0.0.1:1080",
-      "localCoreEndpointDesc": "Chỉ chấp nhận 127.0.0.1 / localhost (SS/VMess/Trojan/VLESS yêu cầu sing-box/lõi xung đột cục bộ).",
+      "localCoreEndpointPlaceholder": "socks5://127.0.0.1:1080\nsocks5://127.0.0.1:1081",
+      "localCoreEndpointDesc": "Mỗi dòng một địa chỉ. Chỉ chấp nhận 127.0.0.1 / localhost (SS/VMess/Trojan/VLESS yêu cầu sing-box/lõi xung đột cục bộ).",
       "routeByProvider": "Định tuyến theo Nhà cung cấp (nhiều lựa chọn)",
       "loadingProviders": "Đang tải danh sách Nhà cung cấp…",
       "autoRefreshInterval": "Khoảng thời gian làm mới tự động (phút)",

src/lib/proxySubscription/coreEndpoint.ts

@@ -1,23 +1,66 @@
 /**
  * Pure, dependency-free validation of a subscription's local proxy-core
- * endpoint.
+ * endpoints (one per line).
  *
  * Extracted from `subscriptionService.isLocalCoreEndpointAllowed` so the
  * security gate is unit-testable without the full DB / Next.js stack.
  */
 
+import { redactSubscriptionUrl } from "./url";
+
 export const ALLOWED_LOCAL_CORE_HOSTS = new Set<string>(["127.0.0.1", "::1", "localhost"]);
 
 /** Only these URL schemes denote a usable local proxy-core endpoint. */
 export const ALLOWED_CORE_SCHEMES = new Set<string>(["http:", "https:", "socks5:"]);
 
+/**
+ * Split a `localCoreEndpoint` field into one entry per line.
+ *
+ * Newline is the only separator (a comma is legitimate inside userinfo and
+ * must survive). Each line is trimmed, blank lines are dropped, order is
+ * preserved. Never throws.
+ */
+export function parseLocalCoreEndpoints(field: string | null): string[] {
+  if (!field) return [];
+  return field
+    .split("\n")
+    .map((line) => line.trim())
+    .filter((line) => line.length > 0);
+}
+
+/**
+ * Redact a raw core entry for the subscription `detail` string.
+ *
+ * `redactSubscriptionUrl` returns non-parseable input unchanged, which could
+ * leak `user:pass` into the stored error column — so strip the userinfo part
+ * as a fallback. Only an `@` inside the authority section (before the first
+ * `/`, `?` or `#` after the scheme) counts: a `@` later in the path is kept
+ * verbatim for diagnostics. Never throws.
+ */
+export function redactCoreEntryForDetail(entry: string): string {
+  try {
+    const redacted = redactSubscriptionUrl(entry);
+    if (redacted !== entry) return redacted;
+    const schemeEnd = entry.indexOf("://");
+    const prefix = schemeEnd >= 0 ? entry.slice(0, schemeEnd + 3) : "";
+    const rest = schemeEnd >= 0 ? entry.slice(schemeEnd + 3) : entry;
+    const delim = rest.search(/[\/?#]/);
+    const authority = delim >= 0 ? rest.slice(0, delim) : rest;
+    const at = authority.lastIndexOf("@");
+    if (at >= 0) return prefix + rest.slice(at + 1);
+    return entry;
+  } catch {
+    return entry;
+  }
+}
+
 /**
  * Whether `endpoint` is an acceptable local proxy-core address.
  *
  * Only loopback hosts over a proxy scheme (http/https/socks5) are permitted.
- * A subscription's `localCoreEndpoint` becomes the single SOCKS5/HTTP address
- * OmniRoute routes SS/VMess/Trojan/VLESS/etc. traffic through, so it must
- * never point at a remote host — and a non-proxy scheme (file:/ftp:/…) is
+ * A subscription's `localCoreEndpoint` lists the SOCKS5/HTTP addresses
+ * OmniRoute routes SS/VMess/Trojan/VLESS/etc. traffic through, so each entry
+ * must never point at a remote host — and a non-proxy scheme (file:/ftp:/…) is
  * meaningless and rejected.
  */
 export function isLocalCoreEndpointAllowed(endpoint: string | null): boolean {

src/lib/proxySubscription/subscriptionService.ts

@@ -32,7 +32,12 @@ import {
 } from "../db/proxies";
 import { bumpProxyConfigGeneration } from "../db/settings";
 import { isSubscriptionDue } from "./due";
-import { isLocalCoreEndpointAllowed } from "./coreEndpoint";
+import {
+  isLocalCoreEndpointAllowed,
+  parseLocalCoreEndpoints,
+  redactCoreEntryForDetail,
+} from "./coreEndpoint";
+import { isProxyReachable } from "../proxyHealth";
 import { resolveTargetScopes } from "./scopes";
 import {
   isSubscriptionFetchUrlAllowed,
@@ -408,6 +413,22 @@ async function keepOwnedSyncedRow(
   }
 }
 
+/**
+ * Build the URL the reachability probe dials for a validated core entry.
+ *
+ * The probe resolves a missing port via its own scheme default (socks5→1080)
+ * while the registry row uses the upsert rule (https→443, else 8080) — so the
+ * probe must carry the row's effective port explicitly, otherwise verdict and
+ * row disagree on port-less entries. Userinfo is preserved for the connection;
+ * the warning `detail` always uses the redacted entry, never this URL.
+ */
+function buildProbeUrl(coreUrl: URL, coreType: string, port: number): string {
+  const auth = coreUrl.username
+    ? `${coreUrl.username}${coreUrl.password ? `:${coreUrl.password}` : ""}@`
+    : "";
+  return `${coreType}://${auth}${coreUrl.hostname.toLowerCase()}:${port}`;
+}
+
 /** Fetch + parse + sync nodes into proxy_registry, then (if enabled) (re)bind. */
 async function syncSubscriptionUnsafe(id: string): Promise<SyncResult> {
   const sub = await getSubscriptionById(id);
@@ -480,39 +501,141 @@ async function syncSubscriptionUnsafe(id: string): Promise<SyncResult> {
       await keepOwnedSyncedRow(upserted, keptIds);
     }
 
-    // needsCore nodes → bind the operator-supplied local core endpoint (single).
+    // needsCore nodes → bind each operator-supplied local core endpoint (one
+    // line of the field becomes its own registry row and pool member).
     if (parsed.needsCore.length > 0) {
-      if (sub.localCoreEndpoint && isLocalCoreEndpointAllowed(sub.localCoreEndpoint)) {
-        try {
-          const coreUrl = new URL(sub.localCoreEndpoint);
+      const entries = parseLocalCoreEndpoints(sub.localCoreEndpoint);
+      if (entries.length === 0) {
+        const nodes = parsed.needsCore
+          .map((n) => `${n.rawProtocol}://${n.host ?? ""}${n.port ? ":" + n.port : ""}`)
+          .join(", ");
+        warning = subscriptionErrorCode("NEEDS_CORE_NOT_CONFIGURED", nodes);
+      } else {
+        // Normalize per entry for keying: gate first, then dedup on
+        // (scheme, host, port, username). The port default follows the upsert
+        // rule below (https→443, else 8080) — not the probe default.
+        const invalid: string[] = [];
+        const unreachable: string[] = [];
+        const collisions: string[] = [];
+        const seenKeys = new Set<string>();
+        const validEntries: Array<{
+          entry: string;
+          probeUrl: string;
+          coreUrl: URL;
+          coreType: string;
+          port: number;
+          username?: string;
+          password?: string;
+        }> = [];
+        for (const entry of entries) {
+          if (!isLocalCoreEndpointAllowed(entry)) {
+            invalid.push(redactCoreEntryForDetail(entry));
+            continue;
+          }
+          let coreUrl: URL;
+          try {
+            coreUrl = new URL(entry);
+          } catch {
+            invalid.push(redactCoreEntryForDetail(entry));
+            continue;
+          }
           const coreType =
             coreUrl.protocol === "https:"
               ? "https"
               : coreUrl.protocol === "socks5:"
                 ? "socks5"
                 : "http";
-          const upserted = await upsertProxy(
-            {
-              name: `${sub.name} (local core)`,
-              type: coreType,
-              host: coreUrl.hostname,
-              port: Number(coreUrl.port) || (coreType === "https" ? 443 : 8080),
-              username: coreUrl.username ? decodeUserinfo(coreUrl.username) : undefined,
-              password: coreUrl.password ? decodeUserinfo(coreUrl.password) : undefined,
-              source: "subscription",
-              subscriptionId: id,
-            },
-            { claimOwnership: false }
-          );
-          await keepOwnedSyncedRow(upserted, keptIds);
-        } catch {
-          warning = subscriptionErrorCode("LOCAL_CORE_ENDPOINT_INVALID");
+          const port = Number(coreUrl.port) || (coreType === "https" ? 443 : 8080);
+          let username = "";
+          let password: string | undefined;
+          try {
+            username = coreUrl.username ? decodeUserinfo(coreUrl.username) : "";
+            password = coreUrl.password ? decodeUserinfo(coreUrl.password) : undefined;
+          } catch {
+            invalid.push(redactCoreEntryForDetail(entry));
+            continue;
+          }
+          const key = `${coreType}://${coreUrl.hostname.toLowerCase()}:${port}:${username}`;
+          if (seenKeys.has(key)) continue;
+          // A scheme collision shares (host, port, username) with an already
+          // accepted entry: the registry keys on that tuple without the scheme,
+          // so a second upsert would silently overwrite the first row's type.
+          // Keep the first entry, report the loser.
+          const ownerKey = `://${coreUrl.hostname.toLowerCase()}:${port}:${username}`;
+          let collided = false;
+          for (const seen of seenKeys) {
+            if (seen.endsWith(ownerKey)) {
+              collided = true;
+              break;
+            }
+          }
+          if (collided) {
+            collisions.push(redactCoreEntryForDetail(entry));
+            continue;
+          }
+          seenKeys.add(key);
+          // Probe the normalized URL, not the raw entry: the probe resolves a
+          // missing port via its own scheme default (socks5→1080) while the
+          // row uses the upsert rule (https→443, else 8080). Probing the raw
+          // entry would test a different port than the row serves.
+          const probeUrl = buildProbeUrl(coreUrl, coreType, port);
+          validEntries.push({
+            entry,
+            probeUrl,
+            coreUrl,
+            coreType,
+            port,
+            username: username || undefined,
+            password,
+          });
+        }
+        // Probe reachability per entry, concurrently: the TCP probes are
+        // independent reads, so they fan out under Promise.all instead of
+        // stacking N sequential timeouts on the periodic sync path. Upserts
+        // stay sequential below. The verdict only feeds the warning detail —
+        // a row is created even when the core is unreachable.
+        const probeVerdicts = await Promise.all(
+          validEntries.map(async (valid) => {
+            try {
+              return await isProxyReachable(valid.probeUrl, undefined, 0);
+            } catch {
+              return false;
+            }
+          })
+        );
+        probeVerdicts.forEach((reachable, i) => {
+          if (!reachable) unreachable.push(redactCoreEntryForDetail(validEntries[i].entry));
+        });
+        for (const valid of validEntries) {
+          try {
+            const upserted = await upsertProxy(
+              {
+                name:
+                  validEntries.length > 1
+                    ? `${sub.name} (local core :${valid.port}/${valid.coreType})`
+                    : `${sub.name} (local core)`,
+                type: valid.coreType,
+                host: valid.coreUrl.hostname,
+                port: valid.port,
+                username: valid.username,
+                password: valid.password,
+                source: "subscription",
+                subscriptionId: id,
+              },
+              { claimOwnership: false }
+            );
+            await keepOwnedSyncedRow(upserted, keptIds);
+          } catch {
+            invalid.push(redactCoreEntryForDetail(valid.entry));
+          }
+        }
+        const parts: string[] = [];
+        if (invalid.length > 0) parts.push(`invalid: ${invalid.join("; ")}`);
+        if (unreachable.length > 0) parts.push(`unreachable: ${unreachable.join("; ")}`);
+        if (collisions.length > 0) parts.push(`key-collision: ${collisions.join("; ")}`);
+        if (parts.length > 0) {
+          warning = subscriptionErrorCode("LOCAL_CORE_ENDPOINT_INVALID", parts.join(" | "));
         }
-      } else {
-        const nodes = parsed.needsCore
-          .map((n) => `${n.rawProtocol}://${n.host ?? ""}${n.port ? ":" + n.port : ""}`)
-          .join(", ");
-        warning = subscriptionErrorCode("NEEDS_CORE_NOT_CONFIGURED", nodes);
       }
     }
 

tests/unit/proxySubscription.coreEndpoint.test.ts

@@ -2,7 +2,12 @@ import test from "node:test";
 import assert from "node:assert/strict";
 
 const mod = await import("../../src/lib/proxySubscription/coreEndpoint.ts");
-const { isLocalCoreEndpointAllowed, ALLOWED_LOCAL_CORE_HOSTS } = mod;
+const {
+  isLocalCoreEndpointAllowed,
+  ALLOWED_LOCAL_CORE_HOSTS,
+  parseLocalCoreEndpoints,
+  redactCoreEntryForDetail,
+} = mod;
 
 test("loopback hosts are allowed", () => {
   assert.equal(isLocalCoreEndpointAllowed("socks5://127.0.0.1:1080"), true);
@@ -27,8 +32,74 @@ test("null / empty / malformed endpoints are rejected", () => {
 });
 
 test("allowed host set is loopback-only", () => {
+  assert.deepEqual([...ALLOWED_LOCAL_CORE_HOSTS].sort(), ["127.0.0.1", "::1", "localhost"].sort());
+});
+
+test("parseLocalCoreEndpoints splits one entry per line", () => {
   assert.deepEqual(
-    [...ALLOWED_LOCAL_CORE_HOSTS].sort(),
-    ["127.0.0.1", "::1", "localhost"].sort()
+    parseLocalCoreEndpoints(
+      "socks5://127.0.0.1:1080\nsocks5://127.0.0.1:1081\nhttp://localhost:2080"
+    ),
+    ["socks5://127.0.0.1:1080", "socks5://127.0.0.1:1081", "http://localhost:2080"]
   );
 });
+
+test("parseLocalCoreEndpoints handles CRLF and drops blank lines", () => {
+  assert.deepEqual(
+    parseLocalCoreEndpoints("socks5://127.0.0.1:1080\r\n\r\n   \r\nsocks5://127.0.0.1:1081\r\n"),
+    ["socks5://127.0.0.1:1080", "socks5://127.0.0.1:1081"]
+  );
+});
+
+test("parseLocalCoreEndpoints trims each line", () => {
+  assert.deepEqual(
+    parseLocalCoreEndpoints("  socks5://127.0.0.1:1080  \n\tsocks5://127.0.0.1:1081\t"),
+    ["socks5://127.0.0.1:1080", "socks5://127.0.0.1:1081"]
+  );
+});
+
+test("parseLocalCoreEndpoints never splits on commas", () => {
+  // A comma is legitimate inside userinfo; it must survive the split.
+  assert.deepEqual(parseLocalCoreEndpoints("socks5://user,name:pass@127.0.0.1:1080"), [
+    "socks5://user,name:pass@127.0.0.1:1080",
+  ]);
+  // Two URLs on one comma-separated line are a single entry (rejected by the
+  // loopback gate later, not by the parser).
+  assert.deepEqual(parseLocalCoreEndpoints("socks5://127.0.0.1:1080, socks5://127.0.0.1:1081"), [
+    "socks5://127.0.0.1:1080, socks5://127.0.0.1:1081",
+  ]);
+});
+
+test("parseLocalCoreEndpoints returns [] for null or blank input", () => {
+  assert.deepEqual(parseLocalCoreEndpoints(null), []);
+  assert.deepEqual(parseLocalCoreEndpoints(""), []);
+  assert.deepEqual(parseLocalCoreEndpoints("   \n  \r\n "), []);
+});
+
+test("parseLocalCoreEndpoints keeps a port-less entry as-is", () => {
+  // The default port (8080 per the upsert rule) is applied in the sync loop.
+  assert.deepEqual(parseLocalCoreEndpoints("socks5://127.0.0.1"), ["socks5://127.0.0.1"]);
+});
+
+test("redactCoreEntryForDetail strips userinfo from parseable entries", () => {
+  const redacted = redactCoreEntryForDetail("socks5://user:pass@127.0.0.1:1080");
+  assert.ok(!redacted.includes("user"), `userinfo leaked: ${redacted}`);
+  assert.ok(!redacted.includes("pass"), `password leaked: ${redacted}`);
+  assert.ok(redacted.includes("127.0.0.1:1080"));
+});
+
+test("redactCoreEntryForDetail leaves entries without userinfo unchanged", () => {
+  assert.equal(redactCoreEntryForDetail("socks5://127.0.0.1:1080"), "socks5://127.0.0.1:1080");
+});
+
+test("redactCoreEntryForDetail strips userinfo even from malformed entries", () => {
+  const redacted = redactCoreEntryForDetail("socks5://user:pass@???");
+  assert.ok(!redacted.includes("user"), `userinfo leaked: ${redacted}`);
+  assert.ok(!redacted.includes("pass"), `password leaked: ${redacted}`);
+});
+
+test("redactCoreEntryForDetail keeps a @ in the path of malformed entries", () => {
+  // No authority userinfo here (no scheme, @ after a path slash): kept
+  // verbatim so the stored detail stays diagnosable.
+  assert.equal(redactCoreEntryForDetail("not-a-url/path@seg"), "not-a-url/path@seg");
+});

tests/unit/proxySubscription.pluralCore.test.ts

@@ -0,0 +1,430 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import http from "node:http";
+import net from "node:net";
+
+// One registry row per valid `localCoreEndpoint` entry. Real loopback TCP
+// servers stand in for reachable cores; a port closed after binding stands in
+// for an unreachable one. No ESM stubbing — every verdict is a real probe.
+
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-sub-plural-core-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+
+const core = await import("../../src/lib/db/core.ts");
+const proxies = await import("../../src/lib/db/proxies.ts");
+const sub = await import("../../src/lib/proxySubscription/index.ts");
+
+function reset() {
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+  fs.mkdirSync(TEST_DATA_DIR, { recursive: true });
+}
+
+test.after(() => {
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+});
+
+function startTcpServer(): Promise<{ port: number; close: () => Promise<void> }> {
+  return new Promise((resolve) => {
+    const srv = net.createServer((sock) => sock.end());
+    srv.listen(0, "127.0.0.1", () => {
+      const addr = srv.address();
+      if (!addr || typeof addr === "string") throw new Error("no addr");
+      resolve({ port: addr.port, close: () => new Promise((r) => srv.close(() => r())) });
+    });
+  });
+}
+
+async function closedPort(): Promise<number> {
+  const srv = await startTcpServer();
+  const port = srv.port;
+  await srv.close();
+  return port;
+}
+
+function startFeedServer(initialBody: string): Promise<{
+  url: string;
+  setBody: (body: string) => void;
+  close: () => Promise<void>;
+}> {
+  let body = initialBody;
+  return new Promise((resolve) => {
+    const srv = http.createServer((_req, res) => {
+      res.writeHead(200, { "Content-Type": "text/plain" });
+      res.end(body);
+    });
+    srv.listen(0, "127.0.0.1", () => {
+      const addr = srv.address();
+      if (!addr || typeof addr === "string") throw new Error("no addr");
+      resolve({
+        url: `http://127.0.0.1:${addr.port}/list`,
+        setBody: (next) => {
+          body = next;
+        },
+        close: () => new Promise((r) => srv.close(() => r())),
+      });
+    });
+  });
+}
+
+const NEEDS_CORE_FEED = "ss://YWVzLTI1Ni1nY206cGFzcw@203.0.113.9:8388#ss-node";
+
+function insertSubscription(id: string, url: string, localCoreEndpoint: string | null) {
+  const now = new Date().toISOString();
+  core
+    .getDbInstance()
+    .prepare(
+      `INSERT INTO proxy_subscriptions
+        (id, name, url, enabled, mode, rule_providers, local_core_endpoint, update_interval_minutes, status, created_at, updated_at)
+       VALUES (?, ?, ?, 1, 'global', NULL, ?, 60, 'empty', ?, ?)`
+    )
+    .run(id, `sub-${id}`, url, localCoreEndpoint, now, now);
+}
+
+function rowsForSub(id: string) {
+  return core
+    .getDbInstance()
+    .prepare("SELECT id, name, host, port FROM proxy_registry WHERE subscription_id = ?")
+    .all(id) as Array<{ id: string; name: string; host: string; port: number }>;
+}
+
+function errorOf(id: string): string | null {
+  const row = core
+    .getDbInstance()
+    .prepare("SELECT error FROM proxy_subscriptions WHERE id = ?")
+    .get(id) as { error: string | null } | undefined;
+  return row?.error ?? null;
+}
+
+function detailOf(id: string): { code?: string; detail?: string } | null {
+  const raw = errorOf(id);
+  if (!raw) return null;
+  try {
+    return JSON.parse(raw);
+  } catch {
+    return null;
+  }
+}
+
+test("single entry keeps the exact legacy name, row and status", async () => {
+  reset();
+  const tcp = await startTcpServer();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription("s1", feed.url, `socks5://127.0.0.1:${tcp.port}`);
+    const result = await sub.syncSubscription("s1");
+    assert.equal(result.status, "ok");
+    assert.equal(result.error, null);
+    const rows = rowsForSub("s1");
+    assert.equal(rows.length, 1);
+    assert.equal(rows[0].name, "sub-s1 (local core)");
+  } finally {
+    await feed.close();
+    await tcp.close();
+  }
+});
+
+test("single gate-rejected entry warns INVALID with a redacted invalid detail", async () => {
+  reset();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription("s1", feed.url, "socks5://user:pass@192.0.2.1:1080");
+    const result = await sub.syncSubscription("s1");
+    assert.equal(result.status, "error");
+    assert.equal(rowsForSub("s1").length, 0);
+    const parsed = detailOf("s1");
+    assert.equal(parsed?.code, "LOCAL_CORE_ENDPOINT_INVALID");
+    assert.ok(parsed?.detail?.includes("invalid:"), `detail: ${parsed?.detail}`);
+    assert.ok(!parsed?.detail?.includes("pass"), `password leaked: ${parsed?.detail}`);
+    assert.ok(!parsed?.detail?.includes("user@"), `userinfo leaked: ${parsed?.detail}`);
+  } finally {
+    await feed.close();
+  }
+});
+
+test("three reachable entries create three rows and three pool members", async () => {
+  reset();
+  const a = await startTcpServer();
+  const b = await startTcpServer();
+  const c = await startTcpServer();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      `socks5://127.0.0.1:${a.port}\nsocks5://127.0.0.1:${b.port}\nsocks5://127.0.0.1:${c.port}`
+    );
+    const result = await sub.syncSubscription("s1");
+    assert.equal(result.status, "ok");
+    const rows = rowsForSub("s1");
+    assert.equal(rows.length, 3);
+    const seen = new Set<number>();
+    for (let i = 0; i < 30; i++) {
+      const r = await proxies.resolveProxyForScopeFromRegistry("global", "");
+      const port = (r as { proxy: { port: number } } | null)?.proxy?.port;
+      if (port) seen.add(port);
+      if (seen.size === 3) break;
+    }
+    assert.ok(seen.has(a.port), `member :${a.port} never visited`);
+    assert.ok(seen.has(b.port), `member :${b.port} never visited`);
+    assert.ok(seen.has(c.port), `member :${c.port} never visited`);
+  } finally {
+    await feed.close();
+    await a.close();
+    await b.close();
+    await c.close();
+  }
+});
+
+test("gate-rejected entry creates no row but warns with a redacted detail", async () => {
+  reset();
+  const a = await startTcpServer();
+  const b = await startTcpServer();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      `socks5://127.0.0.1:${a.port}\nsocks5://user:pass@192.0.2.1:1080\nsocks5://127.0.0.1:${b.port}`
+    );
+    const result = await sub.syncSubscription("s1");
+    assert.equal(result.status, "ok");
+    assert.equal(rowsForSub("s1").length, 2);
+    const parsed = detailOf("s1");
+    assert.equal(parsed?.code, "LOCAL_CORE_ENDPOINT_INVALID");
+    assert.ok(parsed?.detail?.includes("invalid:"), `detail: ${parsed?.detail}`);
+    assert.ok(!parsed?.detail?.includes("pass"), `password leaked: ${parsed?.detail}`);
+    assert.ok(!parsed?.detail?.includes("user@"), `userinfo leaked: ${parsed?.detail}`);
+  } finally {
+    await feed.close();
+    await a.close();
+    await b.close();
+  }
+});
+
+test("unreachable entry still creates its row and is listed as unreachable", async () => {
+  reset();
+  const open = await startTcpServer();
+  const shut = await closedPort();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      `socks5://127.0.0.1:${open.port}\nsocks5://127.0.0.1:${shut}`
+    );
+    const result = await sub.syncSubscription("s1");
+    assert.equal(result.status, "ok");
+    assert.equal(rowsForSub("s1").length, 2);
+    const parsed = detailOf("s1");
+    assert.equal(parsed?.code, "LOCAL_CORE_ENDPOINT_INVALID");
+    assert.ok(parsed?.detail?.includes("unreachable:"), `detail: ${parsed?.detail}`);
+    assert.ok(parsed?.detail?.includes(String(shut)), `detail: ${parsed?.detail}`);
+  } finally {
+    await feed.close();
+    await open.close();
+  }
+});
+
+test("re-sync of an unreachable entry keeps the row id and its assignments", async () => {
+  reset();
+  const shut = await closedPort();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription("s1", feed.url, `socks5://127.0.0.1:${shut}`);
+    await sub.syncSubscription("s1");
+    const before = rowsForSub("s1");
+    assert.equal(before.length, 1);
+    const pooled = core
+      .getDbInstance()
+      .prepare("SELECT COUNT(*) AS n FROM proxy_assignments WHERE proxy_id = ?")
+      .get(before[0].id) as { n: number };
+    assert.equal(pooled.n, 1);
+
+    await sub.syncSubscription("s1");
+
+    const after = rowsForSub("s1");
+    assert.equal(after.length, 1);
+    assert.equal(after[0].id, before[0].id);
+    const pooled2 = core
+      .getDbInstance()
+      .prepare("SELECT COUNT(*) AS n FROM proxy_assignments WHERE proxy_id = ?")
+      .get(before[0].id) as { n: number };
+    assert.equal(pooled2.n, 1);
+  } finally {
+    await feed.close();
+  }
+});
+
+test("removing an entry from the field deletes its row on re-sync", async () => {
+  reset();
+  const a = await startTcpServer();
+  const b = await startTcpServer();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      `socks5://127.0.0.1:${a.port}\nsocks5://127.0.0.1:${b.port}`
+    );
+    await sub.syncSubscription("s1");
+    assert.equal(rowsForSub("s1").length, 2);
+
+    core
+      .getDbInstance()
+      .prepare("UPDATE proxy_subscriptions SET local_core_endpoint = ? WHERE id = ?")
+      .run(`socks5://127.0.0.1:${a.port}`, "s1");
+    await sub.syncSubscription("s1");
+
+    const rows = rowsForSub("s1");
+    assert.equal(rows.length, 1);
+    assert.equal(rows[0].port, a.port);
+  } finally {
+    await feed.close();
+    await a.close();
+    await b.close();
+  }
+});
+
+test("scheme collision on one key creates one row and an exact key-collision prefix", async () => {
+  reset();
+  const tcp = await startTcpServer();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      `http://127.0.0.1:${tcp.port}\nsocks5://127.0.0.1:${tcp.port}`
+    );
+    await sub.syncSubscription("s1");
+    assert.equal(rowsForSub("s1").length, 1);
+    const parsed = detailOf("s1");
+    assert.equal(parsed?.code, "LOCAL_CORE_ENDPOINT_INVALID");
+    assert.ok(parsed?.detail?.includes("key-collision:"), `detail: ${parsed?.detail}`);
+  } finally {
+    await feed.close();
+    await tcp.close();
+  }
+});
+
+test("combined feed lists all three motifs joined in order", async () => {
+  reset();
+  const open = await startTcpServer();
+  const shut = await closedPort();
+  const clash = await startTcpServer();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      [
+        `socks5://127.0.0.1:${open.port}`,
+        "socks5://192.0.2.1:1080",
+        `socks5://127.0.0.1:${shut}`,
+        `http://127.0.0.1:${clash.port}`,
+        `socks5://127.0.0.1:${clash.port}`,
+      ].join("\n")
+    );
+    await sub.syncSubscription("s1");
+    assert.equal(rowsForSub("s1").length, 3);
+    const parsed = detailOf("s1");
+    assert.equal(parsed?.code, "LOCAL_CORE_ENDPOINT_INVALID");
+    const detail = parsed?.detail ?? "";
+    const iInvalid = detail.indexOf("invalid:");
+    const iUnreachable = detail.indexOf("unreachable:");
+    const iCollision = detail.indexOf("key-collision:");
+    assert.ok(iInvalid >= 0 && iUnreachable >= 0 && iCollision >= 0, `detail: ${detail}`);
+    assert.ok(iInvalid < iUnreachable && iUnreachable < iCollision, `detail: ${detail}`);
+    assert.ok(detail.includes(" | "), `detail: ${detail}`);
+  } finally {
+    await feed.close();
+    await open.close();
+    await clash.close();
+  }
+});
+
+test("detail never carries a password or userinfo", async () => {
+  reset();
+  const shut = await closedPort();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      `socks5://operator:s3cret@127.0.0.1:${shut}\nsocks5://intruder:hunter2@192.0.2.1:1080`
+    );
+    await sub.syncSubscription("s1");
+    const raw = errorOf("s1") ?? "";
+    for (const secret of ["s3cret", "hunter2", "operator@", "intruder@"]) {
+      assert.ok(!raw.includes(secret), `leaked ${secret}: ${raw}`);
+    }
+  } finally {
+    await feed.close();
+  }
+});
+
+test("port-less entry probes the row port, not the probe default", async () => {
+  reset();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    // A listener on the upsert-rule port (8080) proves the verdict came from
+    // the row's effective port, not the probe's scheme default (socks5→1080).
+    const probe = await new Promise<{ port: number; close: () => Promise<void> }>(
+      (resolve, reject) => {
+        const srv = net.createServer((sock) => sock.end());
+        srv.on("error", reject);
+        srv.listen(8080, "127.0.0.1", () =>
+          resolve({ port: 8080, close: () => new Promise((r) => srv.close(() => r())) })
+        );
+      }
+    );
+    try {
+      insertSubscription("s1", feed.url, "socks5://127.0.0.1");
+      await sub.syncSubscription("s1");
+      const rows = rowsForSub("s1");
+      assert.equal(rows.length, 1);
+      assert.equal(rows[0].port, 8080);
+      assert.equal(
+        errorOf("s1"),
+        null,
+        `port-less reachable entry must not warn: ${errorOf("s1")}`
+      );
+    } finally {
+      await probe.close();
+    }
+  } finally {
+    await feed.close();
+  }
+});
+
+test("concurrent probes do not stack sequential timeouts", async () => {
+  reset();
+  const a = await startTcpServer();
+  const b = await startTcpServer();
+  const c = await startTcpServer();
+  const feed = await startFeedServer(NEEDS_CORE_FEED);
+  try {
+    insertSubscription(
+      "s1",
+      feed.url,
+      `socks5://127.0.0.1:${a.port}\nsocks5://127.0.0.1:${b.port}\nsocks5://127.0.0.1:${c.port}`
+    );
+    const started = Date.now();
+    await sub.syncSubscription("s1");
+    const elapsed = Date.now() - started;
+    assert.equal(rowsForSub("s1").length, 3);
+    // Three reachable loopback probes fan out: far below one full probe
+    // timeout each, let alone three stacked. Generous bound — flakes only if
+    // the probes ever serialize again.
+    assert.ok(elapsed < 6000, `3 probes took ${elapsed}ms, expected concurrent fan-out`);
+  } finally {
+    await feed.close();
+    await a.close();
+    await b.close();
+    await c.close();
+  }
+});