Compare commits

...
Author SHA1 Message Date
Saoud Rizwan 36d1a6a8b3 fix(cli): queue hub stop connector restarts against the canonical hub url
doctor and update pass targetHubUrl: resolveDefaultCliHubUrl() when
stopping connectors so queued restarts match the hub that will actually
come back, but hub stop defaulted targetHubUrl to the old discovery URL.
If the hub restarted on a different port the queue entries never matched
and the connectors stayed queued forever. Use the same canonical URL in
hub stop.
2026-06-10 15:46:15 -07:00
Saoud Rizwan 7835260289 fix(cli): write connector state and restart queue files owner-only
The restart spec persists raw connector CLI args into the state file and
restart-queue.json, and those args can carry secrets such as Slack and
Telegram bot tokens. writeJsonFile previously created these files with
default permissions while hub discovery records are written 0600. Recreate
the files with mode 0600 and chmod them after writing, matching the
discovery record discipline.
2026-06-10 15:45:16 -07:00
Saoud Rizwan 09f714be3f fix(cli): drain connector restart queue on implicit hub starts
Queued connector restarts previously only ran from explicit 'hub start',
'hub ensure', and 'cline update'. After 'cline doctor fix' stopped and
queued connectors, a user who simply ran a normal command would bring the
hub back through ensureCliHubServer (run-zen, schedule client, connector
adapters) and the connectors stayed down until an explicit hub command.
Drain the queue inside ensureCliHubServer so any CLI path that brings the
hub up also relaunches queued connectors, drop the now-redundant explicit
drain in update, and tell doctor users what happens to queued connectors.
2026-06-10 15:43:38 -07:00
Saoud Rizwan e56f8e2180 fix(cli): harden connector restart queue draining
Claim matched queue entries by rewriting the queue file before launching
connectors so a crash mid-restart or a concurrent drain cannot replay an
entry that already launched. Guard against re-entrant drains within one
process since a restarting connector ensures the hub itself. Treat a
connector run that throws as a failed attempt instead of aborting the
drain and losing claimed entries. Drop entries for connectors that no
longer resolve in the registry, and cap failed restarts at three attempts
so a permanently broken entry cannot be retried on every hub start
forever.
2026-06-10 15:41:01 -07:00
Saoud Rizwan 34040f4dfc test(cli): match pgrep separator arg in stale hub daemon doctor test 2026-06-10 15:30:48 -07:00
Saoud Rizwan d3e52c9265 Merge remote-tracking branch 'origin/bee/one-hub' into review/one-hub-fix-merge-test
# Conflicts:
#	apps/cli/src/commands/hub.test.ts
#	apps/cli/src/commands/hub.ts
2026-06-10 15:25:45 -07:00
Saoud Rizwan f0023509b7 refactor(hub): simplify stale discovery clearing, share capability list
shouldClearStaleHubDiscovery was only ever called with
discoveredVerified=false (the true assignment sits on a return path),
so the expected-hub probe and compatibility check had no effect and the
condition reduced to "a discovery record exists and was not reused".
Replace it with a plain conditional and drop the tests that exercised
unreachable states.

Also move the hub capability list into a typed HUB_CAPABILITIES
constant in @cline/shared next to HubCapabilityName so the server
cannot drift from the type.
2026-06-09 19:10:35 -07:00
Saoud Rizwan 61d3f9c46e fix(hub): retire legacy shared-owner hubs on production startup
Pre-singleton production builds tracked the local hub under the shared
owner discovery path and spawned daemons on random fallback ports. The
production owner context never reads that path, so upgrades would leave
those daemons running indefinitely with no way to reuse or stop them.
Retire the recorded legacy hub (its record carries the auth token and
pid needed for a graceful stop) and clear the legacy record before
resolving the production hub.
2026-06-09 19:10:26 -07:00
Saoud Rizwan 3ecf6b2264 fix(cli): stop pgrep from parsing the hub daemon marker as an option
pgrep treats the "--cline-hub-daemon" pattern as an unknown long option
and exits 2, so doctor never found stale daemons from compiled-binary
installs, which are exactly the processes 'cline doctor fix' is told to
clean up. Pass "--" before the pattern to end option parsing.
2026-06-09 19:10:14 -07:00
Saoud Rizwan 98439ff250 Merge branch 'main' into bee/one-hub 2026-06-09 18:05:46 -07:00
Saoud Rizwan cf51936151 Merge branch 'main' into bee/one-hub 2026-06-09 15:41:02 -07:00
Saoud Rizwan 4077bb4693 fix: require explicit hub port fallback in production 2026-06-09 15:03:55 -07:00
Saoud Rizwan d430b24237 Merge branch 'main' into bee/one-hub 2026-06-09 13:54:04 -07:00
abeatrix fa3893e2a8 test 2026-06-09 12:25:55 -07:00
abeatrix 162734782d fix Polynomial regular expression 2026-06-09 11:13:29 -07:00
abeatrix 1686e291a2 patches 2026-06-08 22:50:11 -07:00
abeatrix b1bf80a961 clean up 2026-06-08 21:26:46 -07:00
abeatrix c3a7a7097c Restored daemon cron startup, made discovery auth tokens required again, and fixed graceful hub stop/restart paths to use the selected production/shared owner context. 2026-06-08 20:46:59 -07:00
abeatrix 36de5e4d8f Merge branch 'bee/one-hub' into bee/one-hub-fix 2026-06-08 19:19:11 -07:00
abeatrix 696b72b708 hasExplicitPort 2026-06-08 19:18:40 -07:00
abeatrix 92c039e202 hub 2026-06-08 19:15:32 -07:00
abeatrix 933ead9d80 Merge branch 'bee/one-hub' into bee/one-hub-fix 2026-06-08 19:05:34 -07:00
abeatrix f43f4fdd01 fix 2026-06-08 18:55:24 -07:00
abeatrix d38a98d3e5 Merge branch 'bee/one-hub' into bee/one-hub-fix 2026-06-08 18:33:46 -07:00
abeatrix 0012aef210 patches 2026-06-08 18:23:21 -07:00
abeatrix c623875b85 feat: Manage connector restarts across hub lifecycle cleanup
## Summary
- Ensure hub stop/update/doctor cleanup stops connectors associated with killed hubs.
- Persist restart metadata for detached connectors so they can be relaunched after the hub comes back.
- Add a connector restart queue that rewrites `--rpc-address` to the new hub URL on restart.
- Extend doctor fix output with connector restart queue counts.

## Context
We found that fixing hub singleton behavior still left two lifecycle gaps:
1. stale/random-port hub daemons could remain after normal hub stop/start flows
2. connectors attached to killed hubs could be left running against dead hub URLs

This change makes hub cleanup restart-aware: connectors tied to killed hubs are terminated first, queued, and restarted when the managed hub starts again.

## Verification
- CLI connector restart unit tests
- hub command tests
- doctor command tests
- CLI typecheck
- whitespace diff check
2026-06-08 18:14:29 -07:00
abeatrix fbe6e3212a feat: Enforce a production singleton Cline Hub
This PR changes local Hub startup/discovery so production uses one stable daemon per user machine instead of silently creating additional hubs on random ports.

Replace resolveSharedHubOwnerContext with resolveProductionHubOwnerContext
across doctor and hub server lifecycle management to scope hub discovery
to the production owner.

Additionally:
- Preserve and propagate auth tokens when retiring incompatible hubs
- Throw a clear error when a compatible hub is already running but its
  discovery record is missing, guiding users to run 'cline doctor fix'
- Gate port fallback behind an explicit allowPortFallback override
- Update tests to mock the new production hub owner context
2026-06-08 17:38:49 -07:00
32 changed files with 2881 additions and 208 deletions
+162 -2
View File
@@ -14,16 +14,30 @@ import { getCliBuildInfo } from "../utils/common";
const {
mockSpawnSync,
mockResolveClineDataDir,
mockResolveProductionHubOwnerContext,
mockResolveSharedHubOwnerContext,
mockReadHubDiscovery,
mockProbeHubServer,
mockClearHubDiscovery,
mockCreateHubServerUrl,
mockEnsureDetachedHubServer,
mockStopLocalHubServerGracefully,
mockStopConnectorsForHubs,
mockEnsureFileExists,
mockResolveHubEndpointOptions,
mockStopAllConnectors,
} = vi.hoisted(() => ({
mockSpawnSync: vi.fn(),
mockResolveClineDataDir: vi.fn(() => "/tmp/cline-data"),
mockResolveProductionHubOwnerContext: vi.fn(() => ({
ownerId: "hub-production",
discoveryPath: path.join(
"/tmp/cline-data",
"locks",
"hub",
"production.json",
),
})),
mockResolveSharedHubOwnerContext: vi.fn(() => ({
ownerId: "hub-owner",
discoveryPath: path.join(
@@ -37,8 +51,22 @@ const {
mockReadHubDiscovery: vi.fn(),
mockProbeHubServer: vi.fn(),
mockClearHubDiscovery: vi.fn(),
mockCreateHubServerUrl: vi.fn(
(host: string, port: number, pathname: string) =>
`ws://${host}:${port}${pathname}`,
),
mockEnsureDetachedHubServer: vi.fn(),
mockStopLocalHubServerGracefully: vi.fn(async () => false),
mockStopConnectorsForHubs: vi.fn(async () => ({
stoppedProcesses: 0,
queuedRestarts: 0,
})),
mockEnsureFileExists: vi.fn(),
mockResolveHubEndpointOptions: vi.fn(() => ({
host: "127.0.0.1",
port: 25466,
pathname: "/hub",
})),
mockStopAllConnectors: vi.fn(async () => ({
stoppedProcesses: 0,
stoppedSessions: 0,
@@ -52,10 +80,14 @@ vi.mock("node:child_process", () => ({
vi.mock("@cline/core", () => ({
resolveClineDataDir: mockResolveClineDataDir,
resolveProductionHubOwnerContext: mockResolveProductionHubOwnerContext,
resolveSharedHubOwnerContext: mockResolveSharedHubOwnerContext,
clearHubDiscovery: mockClearHubDiscovery,
createHubServerUrl: mockCreateHubServerUrl,
ensureDetachedHubServer: mockEnsureDetachedHubServer,
probeHubServer: mockProbeHubServer,
readHubDiscovery: mockReadHubDiscovery,
resolveHubEndpointOptions: mockResolveHubEndpointOptions,
stopLocalHubServerGracefully: mockStopLocalHubServerGracefully,
ensureFileExists: mockEnsureFileExists,
}));
@@ -64,6 +96,10 @@ vi.mock("../connectors/common", () => ({
isProcessRunning: vi.fn(() => false),
}));
vi.mock("../connectors/restart", () => ({
stopConnectorsForHubs: mockStopConnectorsForHubs,
}));
vi.mock("./connect", () => ({
stopAllConnectors: mockStopAllConnectors,
}));
@@ -76,7 +112,20 @@ describe("runDoctorCommand", () => {
afterEach(() => {
vi.clearAllMocks();
mockResolveClineDataDir.mockReturnValue("/tmp/cline-data");
mockResolveProductionHubOwnerContext.mockReturnValue({
ownerId: "hub-production",
discoveryPath: path.join(
"/tmp/cline-data",
"locks",
"hub",
"production.json",
),
});
mockStopLocalHubServerGracefully.mockResolvedValue(false);
mockStopConnectorsForHubs.mockResolvedValue({
stoppedProcesses: 0,
queuedRestarts: 0,
});
mockStopAllConnectors.mockResolvedValue({
stoppedProcesses: 0,
stoppedSessions: 0,
@@ -110,7 +159,8 @@ describe("runDoctorCommand", () => {
command === "pgrep" &&
Array.isArray(args) &&
args[0] === "-fal" &&
args[1] === "/apps/cli/src/index.ts"
args[1] === "--" &&
args[2] === "/apps/cli/src/index.ts"
) {
return {
status: 0,
@@ -252,6 +302,115 @@ describe("runDoctorCommand", () => {
});
});
it("doctor --fix queues active connectors for restart when killing hubs", async () => {
const cwd = "/workspace";
mockReadHubDiscovery.mockResolvedValue({
url: "ws://127.0.0.1:25466/hub",
port: 25466,
pid: 70001,
});
mockProbeHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25466/hub",
port: 25466,
pid: 70001,
});
mockStopLocalHubServerGracefully.mockResolvedValue(true);
mockStopConnectorsForHubs.mockResolvedValue({
stoppedProcesses: 2,
queuedRestarts: 1,
});
mockSpawnSync.mockReturnValue({ status: 1, stdout: "" });
const output: string[] = [];
const code = await runDoctorCommand(
{ cwd, json: true, fix: true },
{
writeln: (text) => {
output.push(text ?? "");
},
writeErr: () => {},
},
);
expect(code).toBe(0);
expect(mockStopConnectorsForHubs).toHaveBeenCalledWith(
expect.any(Array),
expect.any(Object),
{ targetHubUrl: "ws://127.0.0.1:25466/hub" },
);
expect(JSON.parse(output[0] || "")).toMatchObject({
killed: {
connectorProcesses: 2,
connectorProcessesQueuedForRestart: 1,
connectorRestartsQueued: 1,
},
});
});
it("doctor --fix kills stale random-port hub daemons", async () => {
const cwd = "/workspace";
mockReadHubDiscovery.mockResolvedValue({
url: "ws://127.0.0.1:25466/hub",
port: 25466,
pid: 70001,
});
mockProbeHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25466/hub",
port: 25466,
pid: 70001,
});
mockStopLocalHubServerGracefully.mockResolvedValue(true);
mockSpawnSync.mockImplementation((command: string, args?: string[]) => {
if (command === "lsof") {
return {
status: 0,
stdout: "70001\n",
};
}
if (
command === "pgrep" &&
Array.isArray(args) &&
args[0] === "-fal" &&
args[1] === "--" &&
args[2] === "/sdk/packages/core/src/hub/daemon/entry.ts"
) {
return {
status: 0,
stdout: [
"70001 /Users/example/.bun/bin/bun /repo/sdk/packages/core/src/hub/daemon/entry.ts --cwd /workspace --host 127.0.0.1 --port 25466 --pathname /hub",
"70002 /Users/example/.bun/bin/bun /repo/sdk/packages/core/src/hub/daemon/entry.ts --cwd /workspace --host 127.0.0.1 --port 0 --pathname /hub",
].join("\n"),
};
}
return { status: 1, stdout: "" };
});
const killSpy = vi.spyOn(process, "kill").mockImplementation(() => true);
const output: string[] = [];
const code = await runDoctorCommand(
{ cwd, json: true, fix: true },
{
writeln: (text) => {
output.push(text ?? "");
},
writeErr: () => {},
},
);
expect(code).toBe(0);
expect(killSpy).toHaveBeenCalledWith(70002, "SIGKILL");
expect(killSpy).not.toHaveBeenCalledWith(70001, "SIGKILL");
expect(JSON.parse(output[0] || "")).toMatchObject({
before: {
staleHubPids: [70002],
},
killed: {
staleHubDaemons: 1,
},
});
killSpy.mockRestore();
});
it("doctor --fix kills stale code sidecar processes", async () => {
const cwd = "/workspace";
mockReadHubDiscovery.mockResolvedValue(undefined);
@@ -261,7 +420,8 @@ describe("runDoctorCommand", () => {
command === "pgrep" &&
Array.isArray(args) &&
args[0] === "-fal" &&
args[1] === "/src-tauri/bin/code-sidecar"
args[1] === "--" &&
args[2] === "/src-tauri/bin/code-sidecar"
) {
return {
status: 0,
+82 -11
View File
@@ -7,18 +7,21 @@ import {
probeHubServer,
readHubDiscovery,
resolveClineDataDir,
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
stopLocalHubServerGracefully,
} from "@cline/core";
import { formatUptime } from "@cline/shared";
import { formatUptime, resolveClineBuildEnv } from "@cline/shared";
import { Command } from "commander";
import open from "open";
import { isProcessRunning } from "../connectors/common";
import { stopConnectorsForHubs } from "../connectors/restart";
import {
type ActiveConnectorRecord,
listActiveConnectors,
} from "../connectors/status";
import { getCliBuildInfo } from "../utils/common";
import { resolveDefaultCliHubUrl } from "../utils/hub-runtime";
import { c, writeln } from "../utils/output";
import { stopAllConnectors } from "./connect";
@@ -54,6 +57,7 @@ type DoctorStatus = {
hubStartedAt?: string;
hubUptime?: string;
listeningPids: number[];
staleHubPids: number[];
hubStartupLocks: StartupArtifact[];
staleCliPids: number[];
staleSidecarPids: number[];
@@ -77,7 +81,11 @@ function listMatchingProcesses(pattern: string): ProcessRecord[] {
if (process.platform === "win32") {
return [];
}
const result = spawnSync("pgrep", ["-fal", pattern], { encoding: "utf8" });
// "--" stops pgrep's option parsing so patterns that start with dashes
// (e.g. the "--cline-hub-daemon" marker) are treated as patterns.
const result = spawnSync("pgrep", ["-fal", "--", pattern], {
encoding: "utf8",
});
if (result.status !== 0 && result.status !== 1) {
return [];
}
@@ -148,6 +156,25 @@ function listStaleCliPids(): number[] {
.map((record) => record.pid);
}
function listStaleHubPids(currentHubPids: number[]): number[] {
const current = new Set(currentHubPids.filter((pid) => pid > 0));
const patterns = [
"/sdk/packages/core/src/hub/daemon/entry.ts",
"/sdk/packages/core/dist/hub/daemon/entry.js",
"--cline-hub-daemon",
];
const records = new Map<number, ProcessRecord>();
for (const pattern of patterns) {
for (const record of listMatchingProcesses(pattern)) {
if (current.has(record.pid) || /\bpgrep\s+-fal\b/.test(record.command)) {
continue;
}
records.set(record.pid, record);
}
}
return [...records.values()].map((record) => record.pid);
}
function listStaleSidecarPids(): number[] {
const patterns = [
"/apps/examples/desktop-app/sidecar/index.ts",
@@ -235,7 +262,7 @@ function readStartupArtifact(path: string): StartupArtifact | undefined {
}
function listHubStartupLocks(_cwd: string): StartupArtifact[] {
const owner = resolveSharedHubOwnerContext();
const owner = resolveCliHubOwnerContext();
const ownerPath = join(`${owner.discoveryPath}.lock`, "owner.json");
if (!existsSync(ownerPath)) {
return [];
@@ -259,7 +286,7 @@ async function clearHubStartupArtifacts(
_cwd: string,
options?: { clearDiscovery?: boolean },
): Promise<{ startupLocks: number; discovery: number }> {
const owner = resolveSharedHubOwnerContext();
const owner = resolveCliHubOwnerContext();
const startupLocks = listHubStartupLocks(_cwd);
let clearedStartupLocks = 0;
for (const artifact of startupLocks) {
@@ -291,14 +318,25 @@ function formatHubUptimeFromStartedAt(
return formatUptime(Date.now() - timestamp);
}
function resolveCliHubOwnerContext() {
return resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext();
}
async function collectDoctorStatus(cwd: string): Promise<DoctorStatus> {
const owner = resolveSharedHubOwnerContext();
const owner = resolveCliHubOwnerContext();
const discovery = await readHubDiscovery(owner.discoveryPath);
const health = discovery?.url
? await probeHubServer(discovery.url)
? await probeHubServer(discovery.url, { authToken: discovery.authToken })
: undefined;
const current = health ?? discovery;
const hubUptime = formatHubUptimeFromStartedAt(health?.startedAt);
const listeningPids = listListeningPids(current?.port);
const currentHubPids = [
...(current?.pid ? [current.pid] : []),
...listeningPids,
];
return {
cwd,
hubUrl: current?.url,
@@ -306,7 +344,8 @@ async function collectDoctorStatus(cwd: string): Promise<DoctorStatus> {
hubPid: current?.pid,
hubStartedAt: health?.startedAt,
hubUptime,
listeningPids: listListeningPids(current?.port),
listeningPids,
staleHubPids: listStaleHubPids(currentHubPids),
hubStartupLocks: listHubStartupLocks(cwd),
staleCliPids: listStaleCliPids(),
staleSidecarPids: listStaleSidecarPids(),
@@ -388,6 +427,7 @@ export async function runDoctorCommand(
);
writeln(`hub uptime ${c.dim}${before.hubUptime ?? "n/a"}${c.reset}`);
writeln(formatPidList("hub listeners", before.listeningPids));
writeln(formatPidList("stale hub daemons", before.staleHubPids));
writeln(
formatPidList(
"hub startup locks",
@@ -412,6 +452,7 @@ export async function runDoctorCommand(
}
if (
before.listeningPids.length > 0 ||
before.staleHubPids.length > 0 ||
before.staleCliPids.length > 0 ||
before.staleSidecarPids.length > 0
) {
@@ -423,7 +464,9 @@ export async function runDoctorCommand(
}
const gracefullyStoppedHub = before.hubHealthy
? await stopLocalHubServerGracefully().catch(() => false)
? await stopLocalHubServerGracefully(resolveCliHubOwnerContext()).catch(
() => false,
)
: false;
const refreshedAfterGracefulStop = gracefullyStoppedHub
? await collectDoctorStatus(opts.cwd)
@@ -431,13 +474,25 @@ export async function runDoctorCommand(
const killedHub = gracefullyStoppedHub
? 0
: killPids(refreshedAfterGracefulStop.listeningPids);
const staleCliTargets = before.staleCliPids.filter(
const restartAwareStoppedConnectors = await stopConnectorsForHubs(
before.activeConnectors.map((record) => record.hubUrl),
{ writeln: () => {}, writeErr: () => {} },
{ targetHubUrl: resolveDefaultCliHubUrl() },
);
const staleHubTargets = before.staleHubPids.filter(
(pid) => !refreshedAfterGracefulStop.listeningPids.includes(pid),
);
const killedStaleHubs = killPids(staleHubTargets);
const staleCliTargets = before.staleCliPids.filter(
(pid) =>
!refreshedAfterGracefulStop.listeningPids.includes(pid) &&
!staleHubTargets.includes(pid),
);
const killedCli = killPids(staleCliTargets);
const staleSidecarTargets = before.staleSidecarPids.filter(
(pid) =>
!refreshedAfterGracefulStop.listeningPids.includes(pid) &&
!staleHubTargets.includes(pid) &&
!staleCliTargets.includes(pid),
);
const killedSidecars = killPids(staleSidecarTargets);
@@ -459,9 +514,15 @@ export async function runDoctorCommand(
after,
killed: {
hubListeners: killedHub,
staleHubDaemons: killedStaleHubs,
cliProcesses: killedCli,
sidecarProcesses: killedSidecars,
connectorProcesses: stoppedConnectors.stoppedProcesses,
connectorProcesses:
stoppedConnectors.stoppedProcesses +
restartAwareStoppedConnectors.stoppedProcesses,
connectorProcessesQueuedForRestart:
restartAwareStoppedConnectors.queuedRestarts,
connectorRestartsQueued: restartAwareStoppedConnectors.queuedRestarts,
connectorSessions: stoppedConnectors.stoppedSessions,
hubStartupLocks: clearedArtifacts.startupLocks,
hubDiscovery: clearedArtifacts.discovery,
@@ -471,11 +532,20 @@ export async function runDoctorCommand(
return 0;
}
writeln(`killed hub listeners ${c.dim}${killedHub}${c.reset}`);
writeln(`killed stale hub daemons ${c.dim}${killedStaleHubs}${c.reset}`);
writeln(`killed cli processes ${c.dim}${killedCli}${c.reset}`);
writeln(`killed sidecar processes ${c.dim}${killedSidecars}${c.reset}`);
writeln(
`stopped connector processes ${c.dim}${stoppedConnectors.stoppedProcesses}${c.reset}`,
`stopped connector processes ${c.dim}${stoppedConnectors.stoppedProcesses + restartAwareStoppedConnectors.stoppedProcesses}${c.reset}`,
);
writeln(
`queued connector restarts ${c.dim}${restartAwareStoppedConnectors.queuedRestarts}${c.reset}`,
);
if (restartAwareStoppedConnectors.queuedRestarts > 0) {
writeln(
`${c.dim}queued connectors relaunch automatically the next time the hub starts (any cline command, or 'cline hub start')${c.reset}`,
);
}
writeln(
`stopped connector sessions ${c.dim}${stoppedConnectors.stoppedSessions}${c.reset}`,
);
@@ -487,6 +557,7 @@ export async function runDoctorCommand(
);
writeln(`hub healthy after fix: ${after.hubHealthy ? "yes" : "no"}`);
writeln(formatPidList("remaining hub listeners", after.listeningPids));
writeln(formatPidList("remaining stale hub daemons", after.staleHubPids));
writeln(
formatPidList(
"remaining hub startup locks",
+141 -1
View File
@@ -1,22 +1,37 @@
import { describe, expect, it, vi } from "vitest";
import { afterEach, describe, expect, it, vi } from "vitest";
const {
mockClearHubDiscovery,
mockEnsureDetachedHubServer,
mockProbeHubServer,
mockReadHubDiscovery,
mockResolveProductionHubOwnerContext,
mockResolveSharedHubOwnerContext,
mockRestartQueuedConnectorsForHub,
mockStopLocalHubServerGracefully,
mockStopConnectorsForHubs,
} = vi.hoisted(() => ({
mockClearHubDiscovery: vi.fn(),
mockEnsureDetachedHubServer: vi.fn(),
mockProbeHubServer: vi.fn(),
mockReadHubDiscovery: vi.fn(),
mockResolveProductionHubOwnerContext: vi.fn(() => ({
ownerId: "hub-production",
discoveryPath: "/tmp/cline-data/locks/hub/production.json",
})),
mockResolveSharedHubOwnerContext: vi.fn(() => ({
ownerId: "hub-owner",
discoveryPath: "/tmp/cline-data/locks/hub/owners/hub-owner.json",
})),
mockRestartQueuedConnectorsForHub: vi.fn(async () => ({
restarted: 0,
remaining: 0,
})),
mockStopLocalHubServerGracefully: vi.fn(),
mockStopConnectorsForHubs: vi.fn(async () => ({
stoppedProcesses: 0,
queuedRestarts: 0,
})),
}));
vi.mock("@cline/core", () => ({
@@ -24,13 +39,34 @@ vi.mock("@cline/core", () => ({
ensureDetachedHubServer: mockEnsureDetachedHubServer,
probeHubServer: mockProbeHubServer,
readHubDiscovery: mockReadHubDiscovery,
resolveProductionHubOwnerContext: mockResolveProductionHubOwnerContext,
resolveSharedHubOwnerContext: mockResolveSharedHubOwnerContext,
stopLocalHubServerGracefully: mockStopLocalHubServerGracefully,
}));
vi.mock("../connectors/restart", () => ({
restartQueuedConnectorsForHub: mockRestartQueuedConnectorsForHub,
stopConnectorsForHubs: mockStopConnectorsForHubs,
}));
vi.mock("../utils/hub-runtime", () => ({
resolveDefaultCliHubUrl: () => "ws://127.0.0.1:25463/hub",
}));
import { createHubCommand } from "./hub";
const originalBuildEnv = process.env.CLINE_BUILD_ENV;
describe("createHubCommand", () => {
afterEach(() => {
vi.clearAllMocks();
if (originalBuildEnv === undefined) {
delete process.env.CLINE_BUILD_ENV;
} else {
process.env.CLINE_BUILD_ENV = originalBuildEnv;
}
});
it("includes uptime in hub status output", async () => {
vi.spyOn(Date, "now").mockReturnValue(
new Date("2026-01-01T00:01:05.000Z").getTime(),
@@ -73,4 +109,108 @@ describe("createHubCommand", () => {
uptime: "1m 5s",
});
});
it("queues associated connectors on stop", async () => {
mockReadHubDiscovery.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
port: 25463,
pid: 50174,
});
mockStopLocalHubServerGracefully.mockResolvedValue(true);
mockStopConnectorsForHubs.mockResolvedValue({
stoppedProcesses: 2,
queuedRestarts: 2,
});
const output: string[] = [];
let exitCode = 0;
const cmd = createHubCommand(
{
writeln: (text) => {
output.push(text ?? "");
},
writeErr: () => {},
},
(code) => {
exitCode = code;
},
);
await cmd.parseAsync(["stop"], { from: "user" });
expect(exitCode).toBe(0);
expect(mockStopConnectorsForHubs).toHaveBeenCalledWith(
["ws://127.0.0.1:25463/hub"],
expect.any(Object),
{ targetHubUrl: "ws://127.0.0.1:25463/hub" },
);
expect(JSON.parse(output.at(-1) || "")).toMatchObject({
stopped: true,
stoppedConnectorProcesses: 2,
queuedConnectorRestarts: 2,
});
});
it("restarts queued connectors on start", async () => {
mockEnsureDetachedHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
authToken: "token",
});
const output: string[] = [];
let exitCode = 0;
const cmd = createHubCommand(
{
writeln: (text) => {
output.push(text ?? "");
},
writeErr: () => {},
},
(code) => {
exitCode = code;
},
);
await cmd.parseAsync(["start"], { from: "user" });
expect(exitCode).toBe(0);
expect(mockRestartQueuedConnectorsForHub).toHaveBeenCalledWith(
"ws://127.0.0.1:25463/hub",
expect.any(Object),
);
expect(output.at(-1)).toBe("ws://127.0.0.1:25463/hub");
});
it("passes the selected owner to graceful stop", async () => {
process.env.CLINE_BUILD_ENV = "development";
mockReadHubDiscovery.mockResolvedValue({
url: "ws://127.0.0.1:25466/hub",
port: 25466,
pid: 50174,
});
mockStopLocalHubServerGracefully.mockResolvedValue(true);
const output: string[] = [];
let exitCode = 0;
const cmd = createHubCommand(
{
writeln: (text) => {
output.push(text ?? "");
},
writeErr: () => {},
},
(code) => {
exitCode = code;
},
);
await cmd.parseAsync(["stop"], { from: "user" });
expect(exitCode).toBe(0);
expect(mockStopLocalHubServerGracefully).toHaveBeenCalledWith({
ownerId: "hub-owner",
discoveryPath: "/tmp/cline-data/locks/hub/owners/hub-owner.json",
});
expect(JSON.parse(output[0] || "")).toMatchObject({ stopped: true });
});
});
+45 -10
View File
@@ -3,23 +3,45 @@ import {
ensureDetachedHubServer,
probeHubServer,
readHubDiscovery,
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
stopLocalHubServerGracefully,
} from "@cline/core";
import { formatUptime } from "@cline/shared";
import { formatUptime, resolveClineBuildEnv } from "@cline/shared";
import { Command } from "commander";
import {
restartQueuedConnectorsForHub,
stopConnectorsForHubs,
} from "../connectors/restart";
import { resolveDefaultCliHubUrl } from "../utils/hub-runtime";
interface HubCommandIo {
writeln: (text?: string) => void;
writeErr: (text: string) => void;
}
async function stopHubServer(_workspaceRoot: string): Promise<boolean> {
const owner = resolveSharedHubOwnerContext();
async function stopHubServer(
_workspaceRoot: string,
io: HubCommandIo,
): Promise<{
stopped: boolean;
stoppedConnectorProcesses: number;
queuedConnectorRestarts: number;
}> {
const owner = resolveCliHubOwnerContext();
const discovery = await readHubDiscovery(owner.discoveryPath);
if (await stopLocalHubServerGracefully()) {
const stoppedConnectors = discovery?.url
? await stopConnectorsForHubs([discovery.url], io, {
targetHubUrl: resolveDefaultCliHubUrl(),
})
: { stoppedProcesses: 0, queuedRestarts: 0 };
if (await stopLocalHubServerGracefully(owner)) {
await clearHubDiscovery(owner.discoveryPath);
return true;
return {
stopped: true,
stoppedConnectorProcesses: stoppedConnectors.stoppedProcesses,
queuedConnectorRestarts: stoppedConnectors.queuedRestarts,
};
}
const pid = discovery?.pid;
if (pid) {
@@ -30,7 +52,11 @@ async function stopHubServer(_workspaceRoot: string): Promise<boolean> {
}
}
await clearHubDiscovery(owner.discoveryPath);
return !!pid;
return {
stopped: !!pid,
stoppedConnectorProcesses: stoppedConnectors.stoppedProcesses,
queuedConnectorRestarts: stoppedConnectors.queuedRestarts,
};
}
function formatHubUptimeFromStartedAt(
@@ -46,6 +72,12 @@ function formatHubUptimeFromStartedAt(
return formatUptime(Date.now() - timestamp);
}
function resolveCliHubOwnerContext() {
return resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext();
}
export function createHubCommand(
io: HubCommandIo,
setExitCode: (code: number) => void,
@@ -89,6 +121,7 @@ export function createHubCommand(
port: opts.port,
pathname: opts.pathname,
});
await restartQueuedConnectorsForHub(url, io);
io.writeln(url);
}),
);
@@ -106,16 +139,19 @@ export function createHubCommand(
port: opts.port,
pathname: opts.pathname,
});
await restartQueuedConnectorsForHub(url, io);
io.writeln(url);
}),
);
hub.command("status").action(
action(async () => {
const owner = resolveSharedHubOwnerContext();
const owner = resolveCliHubOwnerContext();
const discovery = await readHubDiscovery(owner.discoveryPath);
const health = discovery?.url
? await probeHubServer(discovery.url)
? await probeHubServer(discovery.url, {
authToken: discovery.authToken,
})
: undefined;
const uptime = formatHubUptimeFromStartedAt(health?.startedAt);
io.writeln(
@@ -133,8 +169,7 @@ export function createHubCommand(
hub.command("stop").action(
action(async () => {
const opts = hub.opts<{ cwd: string }>();
const stopped = await stopHubServer(opts.cwd);
io.writeln(JSON.stringify({ stopped }));
io.writeln(JSON.stringify(await stopHubServer(opts.cwd, io)));
}),
);
+67
View File
@@ -7,10 +7,14 @@ import {
checkForUpdates,
getInstallationInfo,
PackageManager,
resolveCliHubOwnerContext,
withMinimumReleaseAgeBypass,
} from "./update";
const originalArgv = [...process.argv];
const originalBuildEnv = process.env.CLINE_BUILD_ENV;
const originalDataDir = process.env.CLINE_DATA_DIR;
const originalHubDiscoveryPath = process.env.CLINE_HUB_DISCOVERY_PATH;
const originalWrapperPath = process.env.CLINE_WRAPPER_PATH;
const originalGlobalSettingsPath = process.env.CLINE_GLOBAL_SETTINGS_PATH;
const originalIsDev = process.env.IS_DEV;
@@ -32,6 +36,21 @@ function createTempFile(pathSuffix: string): string {
describe("getInstallationInfo", () => {
afterEach(() => {
process.argv = [...originalArgv];
if (originalBuildEnv === undefined) {
delete process.env.CLINE_BUILD_ENV;
} else {
process.env.CLINE_BUILD_ENV = originalBuildEnv;
}
if (originalDataDir === undefined) {
delete process.env.CLINE_DATA_DIR;
} else {
process.env.CLINE_DATA_DIR = originalDataDir;
}
if (originalHubDiscoveryPath === undefined) {
delete process.env.CLINE_HUB_DISCOVERY_PATH;
} else {
process.env.CLINE_HUB_DISCOVERY_PATH = originalHubDiscoveryPath;
}
if (originalWrapperPath === undefined) {
delete process.env.CLINE_WRAPPER_PATH;
} else {
@@ -96,6 +115,21 @@ describe("getInstallationInfo", () => {
describe("auto update settings", () => {
afterEach(() => {
process.argv = [...originalArgv];
if (originalBuildEnv === undefined) {
delete process.env.CLINE_BUILD_ENV;
} else {
process.env.CLINE_BUILD_ENV = originalBuildEnv;
}
if (originalDataDir === undefined) {
delete process.env.CLINE_DATA_DIR;
} else {
process.env.CLINE_DATA_DIR = originalDataDir;
}
if (originalHubDiscoveryPath === undefined) {
delete process.env.CLINE_HUB_DISCOVERY_PATH;
} else {
process.env.CLINE_HUB_DISCOVERY_PATH = originalHubDiscoveryPath;
}
if (originalWrapperPath === undefined) {
delete process.env.CLINE_WRAPPER_PATH;
} else {
@@ -153,6 +187,39 @@ describe("auto update settings", () => {
});
});
describe("hub restart owner selection", () => {
afterEach(() => {
if (originalBuildEnv === undefined) {
delete process.env.CLINE_BUILD_ENV;
} else {
process.env.CLINE_BUILD_ENV = originalBuildEnv;
}
if (originalDataDir === undefined) {
delete process.env.CLINE_DATA_DIR;
} else {
process.env.CLINE_DATA_DIR = originalDataDir;
}
if (originalHubDiscoveryPath === undefined) {
delete process.env.CLINE_HUB_DISCOVERY_PATH;
} else {
process.env.CLINE_HUB_DISCOVERY_PATH = originalHubDiscoveryPath;
}
});
it("uses the shared hub owner outside production builds", () => {
process.env.CLINE_BUILD_ENV = "development";
process.env.CLINE_DATA_DIR = "/tmp/cline-update-test-data";
delete process.env.CLINE_HUB_DISCOVERY_PATH;
const owner = resolveCliHubOwnerContext();
expect(owner.discoveryPath).toContain("/locks/hub/owners/");
expect(owner.discoveryPath).not.toBe(
"/tmp/cline-update-test-data/locks/hub/production.json",
);
});
});
describe("withMinimumReleaseAgeBypass", () => {
it("adds the package-manager-specific cooldown bypass", () => {
expect(
+38 -10
View File
@@ -5,11 +5,17 @@ import {
isAutoUpdateEnabledGlobally,
probeHubServer,
readHubDiscovery,
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
stopLocalHubServerGracefully,
} from "@cline/core";
import { resolveClineBuildEnv } from "@cline/shared";
import { version } from "../../package.json";
import { ensureCliHubServer } from "../utils/hub-runtime";
import { stopConnectorsForHubs } from "../connectors/restart";
import {
ensureCliHubServer,
resolveDefaultCliHubUrl,
} from "../utils/hub-runtime";
import { c, writeErr, writeln } from "../utils/output";
import {
getInstalledKanbanVersion,
@@ -269,13 +275,22 @@ export function getPreferredKanbanInstaller(
const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));
export function resolveCliHubOwnerContext() {
return resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext();
}
async function waitForHubToStop(
url: string,
authToken: string | undefined,
timeoutMs: number,
): Promise<boolean> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
const check = await probeHubServer(url).catch(() => undefined);
const check = await probeHubServer(url, { authToken }).catch(
() => undefined,
);
if (!check?.url) return true;
await sleep(100);
}
@@ -288,20 +303,32 @@ async function waitForHubToStop(
* clears stale discovery, then re-ensures a fresh instance is spawned.
*/
async function restartHubServerIfRunning(): Promise<void> {
const owner = resolveSharedHubOwnerContext();
const owner = resolveCliHubOwnerContext();
const discovery = await readHubDiscovery(owner.discoveryPath).catch(
() => undefined,
);
const health = discovery?.url
? await probeHubServer(discovery.url).catch(() => undefined)
? await probeHubServer(discovery.url, {
authToken: discovery.authToken,
}).catch(() => undefined)
: undefined;
if (!health?.url) return;
if (!discovery || !health?.url) return;
const pid = discovery?.pid;
writeln(`${c.dim}[hub] restarting server…${c.reset}`);
await stopConnectorsForHubs(
[health.url],
{
writeln: () => {},
writeErr: () => {},
},
{
targetHubUrl: resolveDefaultCliHubUrl(),
},
);
let stopped = await stopLocalHubServerGracefully().catch(() => false);
let stopped = await stopLocalHubServerGracefully(owner).catch(() => false);
if (!stopped && pid) {
try {
process.kill(pid, "SIGTERM");
@@ -310,21 +337,22 @@ async function restartHubServerIfRunning(): Promise<void> {
}
}
stopped = await waitForHubToStop(health.url, 3_000);
stopped = await waitForHubToStop(health.url, discovery.authToken, 3_000);
if (!stopped && pid) {
try {
process.kill(pid, "SIGKILL");
} catch {
// best-effort
}
stopped = await waitForHubToStop(health.url, 2_000);
stopped = await waitForHubToStop(health.url, discovery.authToken, 2_000);
}
await clearHubDiscovery(owner.discoveryPath).catch(() => undefined);
// Re-ensure a fresh hub instance is spawned.
// Re-ensure a fresh hub instance is spawned. ensureCliHubServer also
// drains the connector restart queue for the new hub.
try {
await ensureCliHubServer(process.cwd()); // return value intentionally unused here
await ensureCliHubServer(process.cwd());
writeln(`${c.green}${c.reset} ${c.dim}[hub] server restarted${c.reset}`);
} catch (err) {
writeErr(
+35
View File
@@ -3,6 +3,7 @@ import { join } from "node:path";
import { resolveClineDataDir } from "@cline/core";
import { Command, CommanderError } from "commander";
import {
CLINE_CONNECTOR_RESTART_SPEC_ENV,
isProcessRunning,
readJsonFile,
removeFile,
@@ -13,6 +14,7 @@ import {
import type {
ConnectCommandDefinition,
ConnectIo,
ConnectorRestartSpec,
ConnectStopResult,
} from "./types";
@@ -113,6 +115,19 @@ export abstract class ConnectorBase<Options, State>
}
protected writeStateFile(statePath: string, state: unknown): void {
const restart = this.readRestartSpecFromEnv();
if (
restart &&
state &&
typeof state === "object" &&
!Array.isArray(state)
) {
writeJsonFile(statePath, {
...(state as Record<string, unknown>),
restart,
});
return;
}
writeJsonFile(statePath, state);
}
@@ -120,6 +135,26 @@ export abstract class ConnectorBase<Options, State>
removeFile(statePath);
}
private readRestartSpecFromEnv(): ConnectorRestartSpec | undefined {
const raw = process.env[CLINE_CONNECTOR_RESTART_SPEC_ENV]?.trim();
if (!raw) {
return undefined;
}
try {
const parsed = JSON.parse(raw) as Partial<ConnectorRestartSpec>;
if (
parsed.connector === this.name &&
Array.isArray(parsed.args) &&
parsed.args.every((arg) => typeof arg === "string")
) {
return { connector: parsed.connector, args: parsed.args };
}
} catch {
// Ignore malformed restart metadata from the environment.
}
return undefined;
}
protected removeStaleState(
statePath: string,
readState: (path: string) => State | undefined,
+27 -1
View File
@@ -1,5 +1,6 @@
import { spawn } from "node:child_process";
import {
chmodSync,
closeSync,
existsSync,
openSync,
@@ -15,6 +16,8 @@ import { createCliLoggerAdapter } from "../logging/adapter";
import { logSpawnedProcess } from "../logging/process";
import { resolveCliLaunchSpec } from "../utils/internal-launch";
export const CLINE_CONNECTOR_RESTART_SPEC_ENV = "CLINE_CONNECTOR_RESTART_SPEC";
export function parseBooleanFlag(rawArgs: string[], flag: string): boolean {
return rawArgs.includes(flag);
}
@@ -183,6 +186,8 @@ export function spawnDetachedConnector(
}
const detachedLogFd = tryOpenDetachedLogFd(options?.logPath);
try {
const connectorName =
commandPrefixArgs[0] === "connect" ? commandPrefixArgs[1] : undefined;
const child = spawn(command.launcher, command.childArgs, {
cwd: process.cwd(),
detached: true,
@@ -193,6 +198,14 @@ export function spawnDetachedConnector(
env: {
...withResolvedClineBuildEnv(process.env),
[childEnvKey]: "1",
...(connectorName
? {
[CLINE_CONNECTOR_RESTART_SPEC_ENV]: JSON.stringify({
connector: connectorName,
args: rawArgs,
}),
}
: {}),
},
});
logSpawnedProcess({
@@ -259,7 +272,20 @@ export function readJsonFile<T>(path: string, fallback: T): T {
export function writeJsonFile(path: string, value: unknown): void {
ensureParentDir(path);
writeFileSync(path, JSON.stringify(value, null, 2), "utf8");
// Connector state and the restart queue persist raw CLI args, which can
// include secrets like bot tokens. Recreate the file owner-only, matching
// the discipline used for hub discovery records. The mode option only
// applies on create, so remove any existing file first.
rmSync(path, { force: true });
writeFileSync(path, JSON.stringify(value, null, 2), {
encoding: "utf8",
mode: 0o600,
});
try {
chmodSync(path, 0o600);
} catch {
// Best-effort tightening on filesystems without chmod support.
}
}
export function removeFile(path: string): void {
+517
View File
@@ -0,0 +1,517 @@
import {
existsSync,
mkdirSync,
mkdtempSync,
readFileSync,
rmSync,
statSync,
writeFileSync,
} from "node:fs";
import { tmpdir } from "node:os";
import { dirname, join } from "node:path";
import { afterEach, describe, expect, it, vi } from "vitest";
const { mockResolveClineDataDir, mockGetConnector } = vi.hoisted(() => ({
mockResolveClineDataDir: vi.fn(),
mockGetConnector: vi.fn(),
}));
vi.mock("@cline/core", () => ({
resolveClineDataDir: mockResolveClineDataDir,
ensureParentDir: (path: string) => {
mkdirSync(dirname(path), { recursive: true });
},
}));
vi.mock("./registry", () => ({
getConnector: mockGetConnector,
}));
import {
restartQueuedConnectorsForHub,
stopConnectorsForHubs,
} from "./restart";
describe("connector restart queue", () => {
const tempDirs: string[] = [];
afterEach(() => {
vi.restoreAllMocks();
mockGetConnector.mockReset();
mockResolveClineDataDir.mockReset();
for (const dir of tempDirs.splice(0)) {
rmSync(dir, { recursive: true, force: true });
}
});
it("queues connector restart metadata when stopping connectors for a killed hub", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const statePath = join(dataDir, "connectors", "telegram", "bot.json");
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors", "telegram"), { recursive: true });
writeFileSync(
statePath,
JSON.stringify({
botUsername: "bot",
pid: 12345,
rpcAddress: "ws://127.0.0.1:57648/hub",
startedAt: new Date().toISOString(),
restart: {
connector: "telegram",
args: ["-m", "bot", "--rpc-address", "ws://127.0.0.1:57648/hub"],
},
}),
"utf8",
);
const alive = new Set([12345]);
const killSpy = vi
.spyOn(process, "kill")
.mockImplementation((pid, signal) => {
if (signal === 0 || signal === undefined) {
if (alive.has(Number(pid))) {
return true;
}
throw Object.assign(new Error("missing"), { code: "ESRCH" });
}
alive.delete(Number(pid));
return true;
});
const stopped = await stopConnectorsForHubs(
["ws://127.0.0.1:57648/hub"],
{
writeln: () => {},
writeErr: () => {},
},
{
targetHubUrl: "ws://127.0.0.1:25466/hub",
},
);
expect(stopped).toEqual({ stoppedProcesses: 1, queuedRestarts: 1 });
expect(killSpy).toHaveBeenCalledWith(12345, "SIGTERM");
expect(existsSync(statePath)).toBe(false);
expect(JSON.parse(readFileSync(queuePath, "utf8"))).toMatchObject([
{
connector: "telegram",
hubUrl: "ws://127.0.0.1:57648/hub",
targetHubUrl: "ws://127.0.0.1:25466/hub",
pid: 12345,
},
]);
const run = vi.fn(async () => 0);
mockGetConnector.mockResolvedValue({ name: "telegram", run });
const restarted = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
{ writeln: () => {}, writeErr: () => {} },
);
expect(restarted).toEqual({ restarted: 1, remaining: 0 });
expect(run).toHaveBeenCalledWith(
["-m", "bot", "--rpc-address", "ws://127.0.0.1:25466/hub"],
expect.any(Object),
);
expect(existsSync(queuePath)).toBe(false);
});
it("rewrites equals-form rpc address args when restarting queued connectors", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors"), { recursive: true });
writeFileSync(
queuePath,
JSON.stringify([
{
connector: "telegram",
args: ["-m", "bot", "--rpc-address=ws://127.0.0.1:57648/hub"],
hubUrl: "ws://127.0.0.1:57648/hub",
targetHubUrl: "ws://127.0.0.1:25466/hub",
statePath: join(dataDir, "connectors", "telegram", "bot.json"),
pid: 12345,
stoppedAt: new Date().toISOString(),
},
]),
"utf8",
);
const run = vi.fn(async () => 0);
mockGetConnector.mockResolvedValue({ name: "telegram", run });
const restarted = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
{ writeln: () => {}, writeErr: () => {} },
);
expect(restarted).toEqual({ restarted: 1, remaining: 0 });
expect(run).toHaveBeenCalledWith(
["-m", "bot", "--rpc-address=ws://127.0.0.1:25466/hub"],
expect.any(Object),
);
});
it("only restarts queue entries targeted at the started hub", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors"), { recursive: true });
writeFileSync(
queuePath,
JSON.stringify([
{
connector: "telegram",
args: ["-m", "bot-a"],
hubUrl: "ws://127.0.0.1:57648/hub",
targetHubUrl: "ws://127.0.0.1:25466/hub",
statePath: join(dataDir, "connectors", "telegram", "bot-a.json"),
pid: 12345,
stoppedAt: new Date().toISOString(),
},
{
connector: "telegram",
args: ["-m", "bot-b"],
hubUrl: "ws://127.0.0.1:57649/hub",
targetHubUrl: "ws://127.0.0.1:25467/hub",
statePath: join(dataDir, "connectors", "telegram", "bot-b.json"),
pid: 12346,
stoppedAt: new Date().toISOString(),
},
]),
"utf8",
);
const run = vi.fn(async () => 0);
mockGetConnector.mockResolvedValue({ name: "telegram", run });
const restarted = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
{ writeln: () => {}, writeErr: () => {} },
);
expect(restarted).toEqual({ restarted: 1, remaining: 1 });
expect(run).toHaveBeenCalledTimes(1);
expect(run).toHaveBeenCalledWith(
["-m", "bot-a", "--rpc-address", "ws://127.0.0.1:25466/hub"],
expect.any(Object),
);
expect(JSON.parse(readFileSync(queuePath, "utf8"))).toMatchObject([
{
args: ["-m", "bot-b"],
targetHubUrl: "ws://127.0.0.1:25467/hub",
},
]);
});
it.skipIf(process.platform === "win32")(
"writes the restart queue owner-only",
async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const statePath = join(dataDir, "connectors", "telegram", "bot.json");
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors", "telegram"), { recursive: true });
writeFileSync(
statePath,
JSON.stringify({
pid: 12345,
rpcAddress: "ws://127.0.0.1:57648/hub",
restart: {
connector: "telegram",
args: ["--bot-token", "secret"],
},
}),
"utf8",
);
const alive = new Set([12345]);
vi.spyOn(process, "kill").mockImplementation((pid, signal) => {
if (signal === 0 || signal === undefined) {
if (alive.has(Number(pid))) {
return true;
}
throw Object.assign(new Error("missing"), { code: "ESRCH" });
}
alive.delete(Number(pid));
return true;
});
await stopConnectorsForHubs(["ws://127.0.0.1:57648/hub"], {
writeln: () => {},
writeErr: () => {},
});
expect(statSync(queuePath).mode & 0o777).toBe(0o600);
},
);
it("claims queue entries before launching connectors", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors"), { recursive: true });
writeFileSync(
queuePath,
JSON.stringify([
{
connector: "telegram",
args: ["-m", "bot"],
hubUrl: "ws://127.0.0.1:57648/hub",
targetHubUrl: "ws://127.0.0.1:25466/hub",
statePath: join(dataDir, "connectors", "telegram", "bot.json"),
pid: 12345,
stoppedAt: new Date().toISOString(),
},
]),
"utf8",
);
let queueExistedDuringRun: boolean | undefined;
const run = vi.fn(async () => {
queueExistedDuringRun = existsSync(queuePath);
return 0;
});
mockGetConnector.mockResolvedValue({ name: "telegram", run });
const restarted = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
{ writeln: () => {}, writeErr: () => {} },
);
expect(restarted).toEqual({ restarted: 1, remaining: 0 });
expect(queueExistedDuringRun).toBe(false);
});
it("drops queue entries for unknown connectors", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors"), { recursive: true });
writeFileSync(
queuePath,
JSON.stringify([
{
connector: "renamed-connector",
args: ["-m", "bot"],
hubUrl: "ws://127.0.0.1:57648/hub",
targetHubUrl: "ws://127.0.0.1:25466/hub",
statePath: join(dataDir, "connectors", "telegram", "bot.json"),
pid: 12345,
stoppedAt: new Date().toISOString(),
},
]),
"utf8",
);
mockGetConnector.mockResolvedValue(undefined);
const errors: string[] = [];
const restarted = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
{
writeln: () => {},
writeErr: (text) => {
errors.push(text);
},
},
);
expect(restarted).toEqual({ restarted: 0, remaining: 0 });
expect(existsSync(queuePath)).toBe(false);
expect(errors).toEqual([
'[connect] dropping queued restart for unknown connector "renamed-connector"',
]);
});
it("requeues failed restarts with an attempt count and drops them at the cap", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors"), { recursive: true });
const entry = {
connector: "telegram",
args: ["-m", "bot"],
hubUrl: "ws://127.0.0.1:57648/hub",
targetHubUrl: "ws://127.0.0.1:25466/hub",
statePath: join(dataDir, "connectors", "telegram", "bot.json"),
pid: 12345,
stoppedAt: new Date().toISOString(),
};
writeFileSync(queuePath, JSON.stringify([entry]), "utf8");
const run = vi.fn(async () => 1);
mockGetConnector.mockResolvedValue({ name: "telegram", run });
const io = { writeln: () => {}, writeErr: () => {} };
const first = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
io,
);
expect(first).toEqual({ restarted: 0, remaining: 1 });
expect(JSON.parse(readFileSync(queuePath, "utf8"))).toMatchObject([
{ connector: "telegram", attempts: 1 },
]);
const second = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
io,
);
expect(second).toEqual({ restarted: 0, remaining: 1 });
expect(JSON.parse(readFileSync(queuePath, "utf8"))).toMatchObject([
{ connector: "telegram", attempts: 2 },
]);
const errors: string[] = [];
const third = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
{
writeln: () => {},
writeErr: (text) => {
errors.push(text);
},
},
);
expect(third).toEqual({ restarted: 0, remaining: 0 });
expect(existsSync(queuePath)).toBe(false);
expect(errors).toEqual([
'[connect] dropping queued restart for connector "telegram" after 3 failed attempts',
]);
});
it("requeues entries when the connector run throws", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors"), { recursive: true });
writeFileSync(
queuePath,
JSON.stringify([
{
connector: "telegram",
args: ["-m", "bot"],
hubUrl: "ws://127.0.0.1:57648/hub",
targetHubUrl: "ws://127.0.0.1:25466/hub",
statePath: join(dataDir, "connectors", "telegram", "bot.json"),
pid: 12345,
stoppedAt: new Date().toISOString(),
},
]),
"utf8",
);
const run = vi.fn(async () => {
throw new Error("spawn failed");
});
mockGetConnector.mockResolvedValue({ name: "telegram", run });
const restarted = await restartQueuedConnectorsForHub(
"ws://127.0.0.1:25466/hub",
{ writeln: () => {}, writeErr: () => {} },
);
expect(restarted).toEqual({ restarted: 0, remaining: 1 });
expect(JSON.parse(readFileSync(queuePath, "utf8"))).toMatchObject([
{ connector: "telegram", attempts: 1 },
]);
});
it("keeps state and skips restart queue when connector termination fails", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const statePath = join(dataDir, "connectors", "telegram", "bot.json");
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors", "telegram"), { recursive: true });
writeFileSync(
statePath,
JSON.stringify({
botUsername: "bot",
pid: 12345,
rpcAddress: "ws://127.0.0.1:57648/hub",
startedAt: new Date().toISOString(),
restart: {
connector: "telegram",
args: ["-m", "bot"],
},
}),
"utf8",
);
vi.spyOn(process, "kill").mockImplementation((pid, signal) => {
if (signal === 0 || signal === undefined) {
if (Number(pid) === 12345) {
return true;
}
throw Object.assign(new Error("missing"), { code: "ESRCH" });
}
return true;
});
const errors: string[] = [];
const stopped = await stopConnectorsForHubs(["ws://127.0.0.1:57648/hub"], {
writeln: () => {},
writeErr: (text) => {
errors.push(text);
},
});
expect(stopped).toEqual({ stoppedProcesses: 0, queuedRestarts: 0 });
expect(existsSync(statePath)).toBe(true);
expect(existsSync(queuePath)).toBe(false);
expect(errors).toEqual([
"[connect] failed to stop connector pid=12345 hub=ws://127.0.0.1:57648/hub",
]);
});
it("ignores non-directory entries while scanning connector state", async () => {
const dataDir = mkdtempSync(join(tmpdir(), "connector-restart-test-"));
tempDirs.push(dataDir);
mockResolveClineDataDir.mockReturnValue(dataDir);
const statePath = join(dataDir, "connectors", "telegram", "bot.json");
const queuePath = join(dataDir, "connectors", "restart-queue.json");
mkdirSync(join(dataDir, "connectors", "telegram"), { recursive: true });
writeFileSync(queuePath, "[]", "utf8");
writeFileSync(
statePath,
JSON.stringify({
botUsername: "bot",
pid: 12345,
rpcAddress: "ws://127.0.0.1:57648/hub",
startedAt: new Date().toISOString(),
restart: {
connector: "telegram",
args: ["-m", "bot"],
},
}),
"utf8",
);
const alive = new Set([12345]);
vi.spyOn(process, "kill").mockImplementation((pid, signal) => {
if (signal === 0 || signal === undefined) {
if (alive.has(Number(pid))) {
return true;
}
throw Object.assign(new Error("missing"), { code: "ESRCH" });
}
alive.delete(Number(pid));
return true;
});
const stopped = await stopConnectorsForHubs(["ws://127.0.0.1:57648/hub"], {
writeln: () => {},
writeErr: () => {},
});
expect(stopped).toEqual({ stoppedProcesses: 1, queuedRestarts: 1 });
expect(JSON.parse(readFileSync(queuePath, "utf8"))).toMatchObject([
{
connector: "telegram",
targetHubUrl: "ws://127.0.0.1:57648/hub",
},
]);
});
});
+313
View File
@@ -0,0 +1,313 @@
import { existsSync, readdirSync } from "node:fs";
import { join } from "node:path";
import { resolveClineDataDir } from "@cline/core";
import {
isProcessRunning,
readJsonFile,
removeFile,
terminateProcess,
writeJsonFile,
} from "./common";
import { getConnector } from "./registry";
import type { ConnectIo, ConnectorRestartSpec } from "./types";
type ConnectorStateForRestart = {
statePath: string;
pid: number;
hubUrl: string;
restart?: ConnectorRestartSpec;
};
type QueuedConnectorRestart = ConnectorRestartSpec & {
hubUrl: string;
targetHubUrl: string;
statePath: string;
pid: number;
stoppedAt: string;
attempts?: number;
};
const MAX_RESTART_ATTEMPTS = 3;
export type StopConnectorsForHubsOptions = {
targetHubUrl?: string;
};
export type StopConnectorsForHubsResult = {
stoppedProcesses: number;
queuedRestarts: number;
};
export type RestartQueuedConnectorsResult = {
restarted: number;
remaining: number;
};
function restartQueuePath(): string {
return join(resolveClineDataDir(), "connectors", "restart-queue.json");
}
function normalizeHubUrl(url: string): string {
try {
const parsed = new URL(url.includes("://") ? url : `ws://${url}`);
if (parsed.protocol === "http:") {
parsed.protocol = "ws:";
} else if (parsed.protocol === "https:") {
parsed.protocol = "wss:";
}
parsed.search = "";
parsed.hash = "";
return parsed.toString();
} catch {
return url.trim();
}
}
function readQueue(): QueuedConnectorRestart[] {
const parsed = readJsonFile<unknown>(restartQueuePath(), []);
if (!Array.isArray(parsed)) {
return [];
}
return parsed.filter((entry): entry is QueuedConnectorRestart => {
if (!entry || typeof entry !== "object") {
return false;
}
const record = entry as Partial<QueuedConnectorRestart>;
return (
typeof record.connector === "string" &&
Array.isArray(record.args) &&
record.args.every((arg) => typeof arg === "string") &&
typeof record.hubUrl === "string" &&
typeof record.targetHubUrl === "string" &&
typeof record.statePath === "string" &&
typeof record.pid === "number" &&
typeof record.stoppedAt === "string" &&
(record.attempts === undefined || typeof record.attempts === "number")
);
});
}
function writeQueue(queue: QueuedConnectorRestart[]): void {
if (queue.length === 0) {
removeFile(restartQueuePath());
return;
}
writeJsonFile(restartQueuePath(), queue);
}
function listConnectorStatePaths(): string[] {
const root = join(resolveClineDataDir(), "connectors");
if (!existsSync(root)) {
return [];
}
const paths: string[] = [];
for (const entry of readdirSync(root, { withFileTypes: true })) {
if (!entry.isDirectory()) {
continue;
}
const dir = join(root, entry.name);
try {
for (const name of readdirSync(dir)) {
if (name.endsWith(".json") && !name.endsWith(".threads.json")) {
paths.push(join(dir, name));
}
}
} catch {
// Ignore connector directories that disappear while scanning.
}
}
return paths;
}
function readConnectorStateForRestart(
statePath: string,
): ConnectorStateForRestart | undefined {
const parsed = readJsonFile<Record<string, unknown> | undefined>(
statePath,
undefined,
);
if (!parsed) {
return undefined;
}
const pid = typeof parsed.pid === "number" ? parsed.pid : undefined;
const hubUrl =
typeof parsed.hubUrl === "string"
? parsed.hubUrl
: typeof parsed.rpcAddress === "string"
? parsed.rpcAddress
: undefined;
if (!pid || !hubUrl || !isProcessRunning(pid)) {
return undefined;
}
const restart =
parsed.restart &&
typeof parsed.restart === "object" &&
!Array.isArray(parsed.restart)
? (parsed.restart as Partial<ConnectorRestartSpec>)
: undefined;
return {
statePath,
pid,
hubUrl,
restart:
typeof restart?.connector === "string" &&
Array.isArray(restart.args) &&
restart.args.every((arg) => typeof arg === "string")
? { connector: restart.connector, args: restart.args }
: undefined,
};
}
function queueConnectorRestart(
state: ConnectorStateForRestart,
targetHubUrl: string,
): boolean {
if (!state.restart) {
return false;
}
const queue = readQueue().filter(
(entry) =>
entry.statePath !== state.statePath &&
!(
entry.connector === state.restart?.connector && entry.pid === state.pid
),
);
queue.push({
...state.restart,
hubUrl: state.hubUrl,
targetHubUrl,
statePath: state.statePath,
pid: state.pid,
stoppedAt: new Date().toISOString(),
});
writeQueue(queue);
return true;
}
export async function stopConnectorsForHubs(
hubUrls: string[],
io: ConnectIo,
options: StopConnectorsForHubsOptions = {},
): Promise<StopConnectorsForHubsResult> {
const targetHubUrls = new Set(hubUrls.map(normalizeHubUrl));
if (targetHubUrls.size === 0) {
return { stoppedProcesses: 0, queuedRestarts: 0 };
}
const restartTargetHubUrl = options.targetHubUrl
? normalizeHubUrl(options.targetHubUrl)
: undefined;
let stoppedProcesses = 0;
let queuedRestarts = 0;
for (const statePath of listConnectorStatePaths()) {
const state = readConnectorStateForRestart(statePath);
if (!state || !targetHubUrls.has(normalizeHubUrl(state.hubUrl))) {
continue;
}
if (!(await terminateProcess(state.pid))) {
io.writeErr(
`[connect] failed to stop connector pid=${state.pid} hub=${state.hubUrl}`,
);
continue;
}
stoppedProcesses += 1;
io.writeln(
`[connect] stopped connector pid=${state.pid} hub=${state.hubUrl}`,
);
if (
queueConnectorRestart(
state,
restartTargetHubUrl ?? normalizeHubUrl(state.hubUrl),
)
) {
queuedRestarts += 1;
}
removeFile(statePath);
}
return { stoppedProcesses, queuedRestarts };
}
function withHubRpcAddress(args: string[], hubUrl: string): string[] {
const next = args.filter((arg) => arg !== "-i" && arg !== "--interactive");
for (let index = 0; index < next.length; index += 1) {
if (next[index]?.startsWith("--rpc-address=")) {
next[index] = `--rpc-address=${hubUrl}`;
return next;
}
if (next[index] === "--rpc-address" && next[index + 1]) {
next[index + 1] = hubUrl;
return next;
}
}
return [...next, "--rpc-address", hubUrl];
}
// Restarting a connector re-runs its connect command, which ensures the hub
// and drains this queue again. The guard turns those nested drains into
// no-ops so a queue entry is never picked up twice within one process.
let drainInProgress = false;
export async function restartQueuedConnectorsForHub(
hubUrl: string,
io: ConnectIo,
): Promise<RestartQueuedConnectorsResult> {
if (drainInProgress) {
return { restarted: 0, remaining: readQueue().length };
}
const queue = readQueue();
if (queue.length === 0) {
return { restarted: 0, remaining: 0 };
}
const targetHubUrl = normalizeHubUrl(hubUrl);
const matched: QueuedConnectorRestart[] = [];
const remaining: QueuedConnectorRestart[] = [];
for (const entry of queue) {
if (normalizeHubUrl(entry.targetHubUrl) === targetHubUrl) {
matched.push(entry);
} else {
remaining.push(entry);
}
}
if (matched.length === 0) {
return { restarted: 0, remaining: remaining.length };
}
// Claim matched entries before running them so a crash mid-restart (or a
// concurrent drain in another process) cannot replay entries that already
// launched a connector.
writeQueue(remaining);
let restarted = 0;
const failed: QueuedConnectorRestart[] = [];
drainInProgress = true;
try {
for (const entry of matched) {
const connector = await getConnector(entry.connector);
if (!connector) {
io.writeErr(
`[connect] dropping queued restart for unknown connector "${entry.connector}"`,
);
continue;
}
const exitCode = await connector
.run(withHubRpcAddress(entry.args, hubUrl), io)
.catch(() => 1);
if (exitCode === 0) {
restarted += 1;
continue;
}
const attempts = (entry.attempts ?? 0) + 1;
if (attempts >= MAX_RESTART_ATTEMPTS) {
io.writeErr(
`[connect] dropping queued restart for connector "${entry.connector}" after ${attempts} failed attempts`,
);
continue;
}
failed.push({ ...entry, attempts });
}
} finally {
drainInProgress = false;
}
if (failed.length > 0) {
// Re-read before appending so entries queued while restarting survive.
writeQueue([...readQueue(), ...failed]);
}
return { restarted, remaining: readQueue().length };
}
+5
View File
@@ -8,6 +8,11 @@ export type ConnectStopResult = {
stoppedSessions: number;
};
export type ConnectorRestartSpec = {
connector: string;
args: string[];
};
export interface ConnectCommandDefinition {
name: string;
description: string;
+64
View File
@@ -0,0 +1,64 @@
import { afterEach, describe, expect, it, vi } from "vitest";
const { mockEnsureDetachedHubServer, mockRestartQueuedConnectorsForHub } =
vi.hoisted(() => ({
mockEnsureDetachedHubServer: vi.fn(),
mockRestartQueuedConnectorsForHub: vi.fn(async () => ({
restarted: 0,
remaining: 0,
})),
}));
vi.mock("@cline/core", () => ({
createHubServerUrl: (host: string, port: number, pathname: string) =>
`ws://${host}:${port}${pathname}`,
ensureDetachedHubServer: mockEnsureDetachedHubServer,
resolveDefaultHubHost: () => "127.0.0.1",
resolveDefaultHubPort: () => 25463,
resolveHubEndpointOptions: () => ({
host: "127.0.0.1",
port: 25463,
pathname: "/hub",
}),
}));
vi.mock("../connectors/restart", () => ({
restartQueuedConnectorsForHub: mockRestartQueuedConnectorsForHub,
}));
import { ensureCliHubServer } from "./hub-runtime";
describe("ensureCliHubServer", () => {
afterEach(() => {
vi.clearAllMocks();
});
it("drains the connector restart queue after ensuring the hub", async () => {
mockEnsureDetachedHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
authToken: "token",
});
const resolution = await ensureCliHubServer("/workspace");
expect(resolution.url).toBe("ws://127.0.0.1:25463/hub");
expect(mockRestartQueuedConnectorsForHub).toHaveBeenCalledWith(
"ws://127.0.0.1:25463/hub",
expect.any(Object),
);
});
it("returns the hub resolution even when draining the queue fails", async () => {
mockEnsureDetachedHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
authToken: "token",
});
mockRestartQueuedConnectorsForHub.mockRejectedValueOnce(
new Error("queue unreadable"),
);
const resolution = await ensureCliHubServer("/workspace");
expect(resolution.url).toBe("ws://127.0.0.1:25463/hub");
});
});
+16 -1
View File
@@ -1,10 +1,13 @@
import {
createHubServerUrl,
type DetachedHubResolution,
ensureDetachedHubServer,
type HubEndpointOverrides,
resolveDefaultHubHost,
resolveDefaultHubPort,
resolveHubEndpointOptions,
} from "@cline/core";
import { restartQueuedConnectorsForHub } from "../connectors/restart";
/**
* Build a `host:port` rpc address string that respects the current build
@@ -15,6 +18,11 @@ export function resolveDefaultCliRpcAddress(): string {
return `${resolveDefaultHubHost()}:${resolveDefaultHubPort()}`;
}
export function resolveDefaultCliHubUrl(): string {
const endpoint = resolveHubEndpointOptions();
return createHubServerUrl(endpoint.host, endpoint.port, endpoint.pathname);
}
export function parseHubEndpointOverride(
rawAddress: string | undefined,
): HubEndpointOverrides {
@@ -43,5 +51,12 @@ export async function ensureCliHubServer(
workspaceRoot: string,
endpoint: HubEndpointOverrides = {},
): Promise<DetachedHubResolution> {
return await ensureDetachedHubServer(workspaceRoot, endpoint);
const resolution = await ensureDetachedHubServer(workspaceRoot, endpoint);
// Connectors queued by hub stop/doctor/update cleanup come back as soon
// as any CLI path brings the hub up, not only explicit hub commands.
await restartQueuedConnectorsForHub(resolution.url, {
writeln: () => {},
writeErr: () => {},
}).catch(() => undefined);
return resolution;
}
+6 -2
View File
@@ -8,7 +8,7 @@ import {
Llms,
ProviderSettingsManager,
stopLocalHubServerGracefully,
toHubHealthUrl,
toHubStatusUrl,
} from "@cline/core";
import type { HubUINotifyPayload, SessionRecord } from "@cline/shared";
@@ -460,7 +460,11 @@ async function main(): Promise<void> {
const syncHealthState = async (): Promise<void> => {
try {
const response = await fetch(toHubHealthUrl(hubUrl));
const response = await fetch(toHubStatusUrl(hubUrl), {
headers: hubAuthToken
? { authorization: `Bearer ${hubAuthToken}` }
: undefined,
});
if (!response.ok) {
return;
}
+6 -2
View File
@@ -701,7 +701,9 @@ class CoreChatWebviewController implements vscode.Disposable {
const owner = resolveSharedHubOwnerContext();
if (this.hubUrl) {
const healthy = await probeHubServer(this.hubUrl);
const healthy = await probeHubServer(this.hubUrl, {
authToken: this.hubAuthToken,
});
if (healthy?.url) {
return {
url: rememberRecoverableLocalHubUrl(healthy.url, this.hubAuthToken),
@@ -733,7 +735,9 @@ class CoreChatWebviewController implements vscode.Disposable {
): Promise<HubResolution | undefined> {
const discovery = await readHubDiscovery(discoveryPath);
if (!discovery?.url) return undefined;
const healthy = await probeHubServer(discovery.url);
const healthy = await probeHubServer(discovery.url, {
authToken: discovery.authToken,
});
return healthy?.url
? {
url: rememberRecoverableLocalHubUrl(healthy.url, discovery.authToken),
@@ -46,6 +46,35 @@ describe("resolveHubUrl", () => {
await expect(resolveHubUrl()).resolves.toBe("ws://127.0.0.1:25463/hub");
});
it("uses the shared discovery owner in development builds", async () => {
delete process.env.CLINE_HUB_DISCOVERY_PATH;
process.env.CLINE_DATA_DIR = "/tmp/cline-connect-test-data";
process.env.CLINE_BUILD_ENV = "development";
const readHubDiscovery = vi
.spyOn(await import("../discovery"), "readHubDiscovery")
.mockResolvedValue({
hubId: "hub-test",
protocolVersion: "v1",
authToken: "test-token",
host: "127.0.0.1",
port: 25466,
url: "ws://127.0.0.1:25466/hub",
startedAt: new Date(0).toISOString(),
updatedAt: new Date(0).toISOString(),
});
await expect(resolveHubUrl()).resolves.toBe("ws://127.0.0.1:25466/hub");
const discoveryPath = readHubDiscovery.mock.calls[0]?.[0].replaceAll(
"\\",
"/",
);
expect(discoveryPath).toContain("/locks/hub/owners/");
expect(discoveryPath).not.toBe(
"/tmp/cline-connect-test-data/locks/hub/production.json",
);
});
it("falls back to the default endpoint when no discovery file exists", async () => {
process.env.CLINE_HUB_DISCOVERY_PATH = "/tmp/missing-hub-discovery.json";
vi.spyOn(
+13 -3
View File
@@ -3,12 +3,16 @@ import type {
HubReplyEnvelope,
HubTransportFrame,
} from "@cline/shared";
import { resolveClineBuildEnv } from "@cline/shared";
import { createHubServerUrl, readHubDiscovery } from "../discovery";
import {
type HubEndpointOverrides,
resolveHubEndpointOptions,
} from "../discovery/defaults";
import { resolveSharedHubOwnerContext } from "../discovery/workspace";
import {
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
} from "../discovery/workspace";
export interface HubConnection {
send(envelope: HubCommandEnvelope): Promise<HubReplyEnvelope>;
@@ -68,13 +72,19 @@ function sameHubEndpoint(left: string, right: string): boolean {
return leftUrl.toString() === rightUrl.toString();
}
function resolveDefaultHubOwnerContext() {
return resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext();
}
async function resolveHubUrlAuthToken(url: URL): Promise<string | undefined> {
const queryToken = url.searchParams.get("authToken")?.trim();
url.searchParams.delete("authToken");
if (queryToken) {
return queryToken;
}
const owner = resolveSharedHubOwnerContext();
const owner = resolveDefaultHubOwnerContext();
const discovery = await readHubDiscovery(owner.discoveryPath);
if (discovery?.url && sameHubEndpoint(url.toString(), discovery.url)) {
return discovery.authToken;
@@ -87,7 +97,7 @@ export async function resolveHubUrl(
): Promise<string> {
const endpoint = resolveHubEndpointOptions(overrides);
if (!hasExplicitEndpoint(overrides)) {
const owner = resolveSharedHubOwnerContext();
const owner = resolveDefaultHubOwnerContext();
const discovery = await readHubDiscovery(owner.discoveryPath);
if (discovery?.url) {
return discovery.url;
+151 -5
View File
@@ -546,6 +546,10 @@ describe("NodeHubClient", () => {
spawnDetachedHubServerWithRetry: vi.fn(async () => undefined),
}));
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery-recovery.json",
@@ -697,6 +701,10 @@ describe("NodeHubClient", () => {
spawnDetachedHubServerWithRetry: vi.fn(async () => undefined),
}));
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery-explicit.json",
@@ -764,6 +772,10 @@ describe("resolveCompatibleLocalHubUrl", () => {
it("does not clear discovery on transient probe failure", async () => {
const clearHubDiscoveryMock = vi.fn();
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
@@ -798,9 +810,13 @@ describe("resolveCompatibleLocalHubUrl", () => {
expect(clearHubDiscoveryMock).not.toHaveBeenCalled();
});
it("clears discovery on build mismatch", async () => {
it("keeps discovery on build mismatch when protocol is compatible", async () => {
const clearHubDiscoveryMock = vi.fn();
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
@@ -840,15 +856,19 @@ describe("resolveCompatibleLocalHubUrl", () => {
const { resolveCompatibleLocalHubUrl } = await import(".");
await expect(resolveCompatibleLocalHubUrl()).resolves.toBeUndefined();
expect(clearHubDiscoveryMock).toHaveBeenCalledWith(
"/tmp/hub-discovery.json",
await expect(resolveCompatibleLocalHubUrl()).resolves.toBe(
"ws://127.0.0.1:59999/hub",
);
expect(clearHubDiscoveryMock).not.toHaveBeenCalled();
});
it("clears discovery when a hub omits build metadata", async () => {
it("keeps discovery when a hub omits build metadata but has compatible protocol", async () => {
const clearHubDiscoveryMock = vi.fn();
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
@@ -886,6 +906,57 @@ describe("resolveCompatibleLocalHubUrl", () => {
const { resolveCompatibleLocalHubUrl } = await import(".");
await expect(resolveCompatibleLocalHubUrl()).resolves.toBe(
"ws://127.0.0.1:59999/hub",
);
expect(clearHubDiscoveryMock).not.toHaveBeenCalled();
});
it("clears discovery on protocol mismatch", async () => {
const clearHubDiscoveryMock = vi.fn();
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
}));
vi.doMock("../discovery", async () => {
const actual =
await vi.importActual<typeof import("../discovery")>("../discovery");
return {
...actual,
readHubDiscovery: vi.fn(async () => ({
hubId: "hub-test",
protocolVersion: "v0",
buildId: "old-build",
host: "127.0.0.1",
port: 59999,
url: "ws://127.0.0.1:59999/hub",
startedAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
})),
clearHubDiscovery: vi.fn(async (...args: unknown[]) => {
clearHubDiscoveryMock(...args);
}),
probeHubServer: vi.fn(async () => ({
hubId: "hub-test",
protocolVersion: "v0",
buildId: "old-build",
host: "127.0.0.1",
port: 59999,
url: "ws://127.0.0.1:59999/hub",
startedAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
})),
};
});
const { resolveCompatibleLocalHubUrl } = await import(".");
await expect(resolveCompatibleLocalHubUrl()).resolves.toBeUndefined();
expect(clearHubDiscoveryMock).toHaveBeenCalledWith(
"/tmp/hub-discovery.json",
@@ -914,6 +985,10 @@ describe("resolveCompatibleLocalHubUrl", () => {
spawnDetachedHubServerWithRetry: spawnDetachedHubServerWithRetryMock,
}));
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
@@ -950,6 +1025,73 @@ describe("resolveCompatibleLocalHubUrl", () => {
).toBeLessThan(readHubDiscoveryMock.mock.invocationCallOrder[1]);
});
it("waits on shared discovery after spawning in development builds", async () => {
vi.stubGlobal("WebSocket", MockWebSocket);
const originalBuildEnv = process.env.CLINE_BUILD_ENV;
process.env.CLINE_BUILD_ENV = "development";
const spawnDetachedHubServerWithRetryMock = vi.fn(async () => undefined);
const record = {
hubId: "hub-test",
protocolVersion: "v1",
buildId: "test-build",
authToken: "token",
host: "127.0.0.1",
port: 25466,
url: "ws://127.0.0.1:25466/hub",
startedAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
};
const readHubDiscoveryMock = vi.fn(async (path: string) =>
path === "/tmp/shared-hub-discovery.json" ? record : undefined,
);
vi.doMock("../daemon", () => ({
spawnDetachedHubServerWithRetry: spawnDetachedHubServerWithRetryMock,
}));
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "production",
discoveryPath: "/tmp/production-hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "shared",
discoveryPath: "/tmp/shared-hub-discovery.json",
}),
}));
vi.doMock("../discovery", async () => {
const actual =
await vi.importActual<typeof import("../discovery")>("../discovery");
return {
...actual,
readHubDiscovery: readHubDiscoveryMock,
probeHubServer: vi.fn(async () => record),
clearHubDiscovery: vi.fn(async () => undefined),
};
});
try {
const { ensureCompatibleLocalHubUrl } = await import(".");
await expect(
ensureCompatibleLocalHubUrl({
workspaceRoot: "/tmp/project",
cwd: "/tmp/project",
}),
).resolves.toBe("ws://127.0.0.1:25466/hub");
expect(readHubDiscoveryMock).toHaveBeenCalledWith(
"/tmp/shared-hub-discovery.json",
);
expect(readHubDiscoveryMock).not.toHaveBeenCalledWith(
"/tmp/production-hub-discovery.json",
);
} finally {
if (originalBuildEnv === undefined) {
delete process.env.CLINE_BUILD_ENV;
} else {
process.env.CLINE_BUILD_ENV = originalBuildEnv;
}
}
});
it("does not restart explicit local endpoints after startup timeout", async () => {
const readHubDiscoveryMock = vi.fn(async () => ({
hubId: "hub-test",
@@ -963,6 +1105,10 @@ describe("resolveCompatibleLocalHubUrl", () => {
updatedAt: new Date().toISOString(),
}));
vi.doMock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
}),
resolveSharedHubOwnerContext: () => ({
ownerId: "hub-test",
discoveryPath: "/tmp/hub-discovery.json",
+28 -15
View File
@@ -5,6 +5,8 @@ import {
type HubEventEnvelope,
type HubReplyEnvelope,
type HubTransportFrame,
isHubProtocolCompatible,
resolveClineBuildEnv,
resolveHubCommandTimeoutMs,
} from "@cline/shared";
import {
@@ -17,9 +19,11 @@ import {
type HubOwnerContext,
probeHubServer,
readHubDiscovery,
resolveHubBuildId,
} from "../discovery";
import { resolveSharedHubOwnerContext } from "../discovery/workspace";
import {
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
} from "../discovery/workspace";
type PendingReply = {
resolve: (reply: HubReplyEnvelope) => void;
@@ -31,6 +35,12 @@ type SubscriptionEntry = {
sessionId?: string;
};
function resolveDefaultHubOwnerContext(): HubOwnerContext {
return resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext();
}
type WebSocketLike = {
readyState: number;
send(data: string): void;
@@ -821,7 +831,7 @@ type HubProbeResult =
url: string;
}
| {
status: "unreachable" | "build_mismatch";
status: "unreachable" | "protocol_mismatch";
url: string;
};
@@ -835,18 +845,18 @@ async function probeCompatibleHubUrl(
},
): Promise<HubProbeResult> {
const normalized = normalizeHubWebSocketUrl(url);
const record = await probeHubServer(normalized);
const record = await probeHubServer(normalized, {
authToken: options?.authToken,
});
if (!record) {
return {
status: "unreachable",
url: normalized,
};
}
const buildId = resolveHubBuildId();
const recordBuildId = record.buildId?.trim();
if (!recordBuildId || recordBuildId !== buildId) {
if (!isHubProtocolCompatible(record).compatible) {
return {
status: "build_mismatch",
status: "protocol_mismatch",
url: normalized,
};
}
@@ -973,16 +983,18 @@ export async function resolveCompatibleLocalHubUrl(
return compatible.status === "compatible" ? compatible.url : undefined;
}
const owner = resolveSharedHubOwnerContext();
const owner = resolveDefaultHubOwnerContext();
const record = await readHubDiscovery(owner.discoveryPath);
if (!record?.url) {
return undefined;
}
const compatible = await probeCompatibleHubUrl(record.url);
const compatible = await probeCompatibleHubUrl(record.url, {
authToken: record.authToken,
});
if (compatible.status === "compatible") {
return rememberRecoverableLocalHubUrl(compatible.url, record.authToken);
}
if (compatible.status === "build_mismatch") {
if (compatible.status === "protocol_mismatch") {
await clearHubDiscovery(owner.discoveryPath).catch(() => undefined);
}
return undefined;
@@ -1004,7 +1016,7 @@ export async function ensureCompatibleLocalHubUrl(
if (options.endpoint?.trim()) {
return undefined;
}
const owner = resolveSharedHubOwnerContext();
const owner = resolveDefaultHubOwnerContext();
await spawnDetachedHubServerWithRetry(options.workspaceRoot ?? process.cwd());
return await waitForCompatibleHubUrl(owner);
}
@@ -1032,8 +1044,9 @@ export async function requestHubShutdown(
return response.ok;
}
export async function stopLocalHubServerGracefully(): Promise<boolean> {
const owner = resolveSharedHubOwnerContext();
export async function stopLocalHubServerGracefully(
owner: HubOwnerContext = resolveDefaultHubOwnerContext(),
): Promise<boolean> {
const discovery = await readHubDiscovery(owner.discoveryPath);
if (!discovery?.url) {
return false;
@@ -1060,7 +1073,7 @@ export async function restartLocalHubIfIdleAfterStartupTimeout(options: {
if (!isRecoverableLocalHubUrl(options.url)) {
return undefined;
}
const owner = resolveSharedHubOwnerContext();
const owner = resolveDefaultHubOwnerContext();
const discovery = await readHubDiscovery(owner.discoveryPath);
if (!discovery?.url || !sameNormalizedHubUrl(discovery.url, options.url)) {
return undefined;
@@ -0,0 +1,119 @@
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, it, vi } from "vitest";
const {
mockCreateLocalHubScheduleRuntimeHandlers,
mockInitVcr,
mockResolveHubEndpointOptions,
mockResolveProductionHubOwnerContext,
mockResolveSharedHubOwnerContext,
mockStartHubWebSocketServer,
} = vi.hoisted(() => ({
mockCreateLocalHubScheduleRuntimeHandlers: vi.fn(() => ({
startSession: vi.fn(),
sendSession: vi.fn(),
stopSession: vi.fn(),
abortSession: vi.fn(),
})),
mockInitVcr: vi.fn(),
mockResolveHubEndpointOptions: vi.fn(
(options: { host?: string; port?: number; pathname?: string }) => ({
host: options.host ?? "127.0.0.1",
port: options.port ?? 25463,
pathname: options.pathname ?? "/hub",
}),
),
mockResolveProductionHubOwnerContext: vi.fn(() => ({
ownerId: "production",
discoveryPath: "/tmp/cline-data/locks/hub/production.json",
})),
mockResolveSharedHubOwnerContext: vi.fn(() => ({
ownerId: "shared",
discoveryPath: "/tmp/cline-data/locks/hub/owners/shared.json",
})),
mockStartHubWebSocketServer: vi.fn(async () => ({
close: vi.fn(async () => undefined),
})),
}));
vi.mock("@cline/shared", () => ({
initVcr: mockInitVcr,
resolveClineBuildEnv: () => "production",
}));
vi.mock("../daemon/runtime-handlers", () => ({
createLocalHubScheduleRuntimeHandlers:
mockCreateLocalHubScheduleRuntimeHandlers,
}));
vi.mock("../discovery/defaults", () => ({
resolveHubEndpointOptions: mockResolveHubEndpointOptions,
}));
vi.mock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: mockResolveProductionHubOwnerContext,
resolveSharedHubOwnerContext: mockResolveSharedHubOwnerContext,
}));
vi.mock("../server", () => ({
startHubWebSocketServer: mockStartHubWebSocketServer,
}));
const originalArgv = [...process.argv];
const originalCwd = process.cwd();
describe("hub daemon entry", () => {
const tempDirs: string[] = [];
afterEach(() => {
process.argv = [...originalArgv];
process.chdir(originalCwd);
vi.restoreAllMocks();
vi.resetModules();
mockCreateLocalHubScheduleRuntimeHandlers.mockClear();
mockInitVcr.mockClear();
mockResolveHubEndpointOptions.mockClear();
mockResolveProductionHubOwnerContext.mockClear();
mockResolveSharedHubOwnerContext.mockClear();
mockStartHubWebSocketServer.mockClear();
for (const dir of tempDirs.splice(0)) {
rmSync(dir, { recursive: true, force: true });
}
});
it("starts the daemon with cron options for the daemon workspace root", async () => {
const cwd = mkdtempSync(join(tmpdir(), "cline-hub-entry-test-"));
tempDirs.push(cwd);
process.argv = [
"node",
"entry.js",
"--cwd",
cwd,
"--host",
"127.0.0.1",
"--port",
"30000",
"--pathname",
"/hub",
];
vi.spyOn(process, "on").mockImplementation(() => process);
await import("./entry");
await vi.waitFor(() => {
expect(mockStartHubWebSocketServer).toHaveBeenCalled();
});
expect(mockStartHubWebSocketServer).toHaveBeenCalledWith(
expect.objectContaining({
host: "127.0.0.1",
port: 30000,
pathname: "/hub",
owner: expect.objectContaining({ ownerId: "production" }),
cronOptions: { workspaceRoot: cwd },
}),
);
expect(mockCreateLocalHubScheduleRuntimeHandlers).toHaveBeenCalledOnce();
});
});
+9 -3
View File
@@ -1,8 +1,11 @@
import { AgentRuntimeAbortError } from "@cline/agents";
import { initVcr } from "@cline/shared";
import { initVcr, resolveClineBuildEnv } from "@cline/shared";
import { createLocalHubScheduleRuntimeHandlers } from "../daemon/runtime-handlers";
import { resolveHubEndpointOptions } from "../discovery/defaults";
import { resolveSharedHubOwnerContext } from "../discovery/workspace";
import {
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
} from "../discovery/workspace";
import { startHubWebSocketServer } from "../server";
initVcr(process.env.CLINE_VCR);
@@ -62,7 +65,10 @@ async function main(): Promise<void> {
host: endpoint.host,
port: endpoint.port,
pathname: endpoint.pathname,
owner: resolveSharedHubOwnerContext(),
owner:
resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext(),
runtimeHandlers: createLocalHubScheduleRuntimeHandlers(),
cronOptions: { workspaceRoot: options.cwd },
});
+279 -55
View File
@@ -7,6 +7,7 @@ const {
openSync,
rememberRecoverableLocalHubUrl,
verifyHubConnection,
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
createHubServerUrl,
clearHubDiscovery,
@@ -24,6 +25,9 @@ const {
openSync: vi.fn(() => 17),
rememberRecoverableLocalHubUrl: vi.fn((url: string) => url),
verifyHubConnection: vi.fn(),
resolveProductionHubOwnerContext: vi.fn(() => ({
discoveryPath: "/tmp/hub-discovery.json",
})),
resolveSharedHubOwnerContext: vi.fn(() => ({
discoveryPath: "/tmp/hub-discovery.json",
})),
@@ -57,6 +61,9 @@ vi.mock("@cline/shared", () => ({
CLINE_RUN_AS_HUB_DAEMON_ENV,
CLINE_HUB_PORT: 25463,
CLINE_HUB_DEV_PORT: 25466,
isHubProtocolCompatible: (record: { protocolVersion?: string }) => ({
compatible: record.protocolVersion === "v1",
}),
isHubDaemonProcess: (env: NodeJS.ProcessEnv = process.env) =>
env[CLINE_RUN_AS_HUB_DAEMON_ENV] === "1",
resolveClineBuildEnv: () => "production",
@@ -70,6 +77,7 @@ vi.mock("../client", () => ({
}));
vi.mock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
}));
@@ -88,6 +96,21 @@ describe("ensureDetachedHubServer", () => {
beforeEach(() => {
delete process.env[CLINE_RUN_AS_HUB_DAEMON_ENV];
spawn.mockReset();
spawn.mockImplementation(() => ({ unref: vi.fn() }));
closeSync.mockReset();
mkdirSync.mockReset();
openSync.mockReset();
openSync.mockImplementation(() => 17);
rememberRecoverableLocalHubUrl.mockReset();
rememberRecoverableLocalHubUrl.mockImplementation((url: string) => url);
verifyHubConnection.mockReset();
clearHubDiscovery.mockReset();
clearHubDiscovery.mockResolvedValue(undefined);
probeHubServer.mockReset();
requestHubShutdown.mockReset();
requestHubShutdown.mockResolvedValue(true);
readHubDiscovery.mockReset();
vi.stubGlobal("fetch", fetchMock);
});
@@ -101,20 +124,16 @@ describe("ensureDetachedHubServer", () => {
}
});
it("lets the daemon bind port 0 when the configured endpoint is occupied", async () => {
it("does not use port 0 for default production startup", async () => {
readHubDiscovery.mockResolvedValue(undefined);
probeHubServer
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
buildId: "current-build",
})
.mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
buildId: "current-build",
});
probeHubServer.mockResolvedValueOnce(undefined).mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
});
verifyHubConnection.mockResolvedValueOnce(true);
readHubDiscovery.mockResolvedValueOnce(undefined).mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
buildId: "current-build",
authToken: "new-token",
});
@@ -129,12 +148,13 @@ describe("ensureDetachedHubServer", () => {
| undefined;
expect(result).toEqual({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
expect(spawn).toHaveBeenCalledOnce();
expect(spawnArgs).toContain("--port");
expect(spawnArgs).toContain("0");
expect(spawnArgs).toContain("25463");
expect(spawnArgs).not.toContain("0");
expect(spawnOptions?.env?.[CLINE_RUN_AS_HUB_DAEMON_ENV]).toBe("1");
});
@@ -153,11 +173,12 @@ describe("ensureDetachedHubServer", () => {
})
.mockImplementationOnce(() => ({ unref: vi.fn() }));
readHubDiscovery.mockResolvedValueOnce(undefined).mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
probeHubServer.mockResolvedValueOnce(undefined).mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
});
verifyHubConnection.mockResolvedValueOnce(true);
@@ -168,7 +189,7 @@ describe("ensureDetachedHubServer", () => {
const result = await pending;
expect(result).toEqual({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
expect(spawn).toHaveBeenCalledTimes(2);
@@ -247,7 +268,42 @@ describe("ensureDetachedHubServer", () => {
}
});
it("does not reuse a healthy hub from a different build", async () => {
it("prewarms on a fallback port when an empty-token hub cannot be retired", async () => {
vi.useFakeTimers();
const kill = vi.spyOn(process, "kill").mockImplementation(() => true);
try {
readHubDiscovery.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
authToken: "",
pid: 12345,
});
probeHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "old-build",
});
const { prewarmDetachedHubServer } = await import(".");
prewarmDetachedHubServer("/workspace", { allowPortFallback: true });
await vi.runAllTimersAsync();
const spawnArgs = ((spawn as unknown as { mock: { calls: unknown[][] } })
.mock.calls[0]?.[1] ?? []) as string[];
expect(requestHubShutdown).toHaveBeenCalledWith(
"ws://127.0.0.1:25463/hub",
"",
);
expect(kill).toHaveBeenCalledWith(12345, "SIGTERM");
expect(spawn).toHaveBeenCalledOnce();
expect(spawnArgs).toContain("--port");
expect(spawnArgs).toContain("0");
} finally {
kill.mockRestore();
vi.useRealTimers();
}
});
it("reuses a protocol-compatible healthy hub from a different build", async () => {
const kill = vi.spyOn(process, "kill").mockImplementation(() => true);
try {
readHubDiscovery
@@ -255,54 +311,215 @@ describe("ensureDetachedHubServer", () => {
url: "ws://127.0.0.1:25463/hub",
authToken: "old-token",
})
.mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
buildId: "current-build",
authToken: "new-token",
});
probeHubServer
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
buildId: "old-build",
pid: 12345,
})
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
buildId: "old-build",
pid: 12345,
})
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
buildId: "current-build",
});
.mockResolvedValueOnce(undefined);
probeHubServer.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "old-build",
pid: 12345,
});
verifyHubConnection.mockResolvedValueOnce(true);
const { ensureDetachedHubServer } = await import(".");
const result = await ensureDetachedHubServer("/workspace");
expect(result).toEqual({
url: "ws://127.0.0.1:5555/hub",
authToken: "new-token",
url: "ws://127.0.0.1:25463/hub",
authToken: "old-token",
});
expect(requestHubShutdown).toHaveBeenCalledWith(
"ws://127.0.0.1:25463/hub",
"old-token",
);
expect(clearHubDiscovery).toHaveBeenCalledWith("/tmp/hub-discovery.json");
expect(clearHubDiscovery.mock.invocationCallOrder[0]).toBeGreaterThan(
probeHubServer.mock.invocationCallOrder[2],
);
expect(kill).toHaveBeenCalledWith(12345, "SIGTERM");
expect(spawn).toHaveBeenCalledOnce();
expect(requestHubShutdown).not.toHaveBeenCalled();
expect(clearHubDiscovery).not.toHaveBeenCalled();
expect(kill).not.toHaveBeenCalled();
expect(spawn).not.toHaveBeenCalled();
expect(verifyHubConnection).toHaveBeenCalledOnce();
} finally {
kill.mockRestore();
}
});
it("does not reuse a healthy hub without build metadata", async () => {
it("retires an existing hub with an empty discovery auth token before starting a replacement", async () => {
const kill = vi.spyOn(process, "kill").mockImplementation(() => true);
try {
readHubDiscovery
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
authToken: "",
pid: 12345,
})
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
probeHubServer
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
});
verifyHubConnection.mockResolvedValueOnce(true);
const { ensureDetachedHubServer } = await import(".");
const result = await ensureDetachedHubServer("/workspace");
expect(result).toEqual({
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
expect(requestHubShutdown).toHaveBeenCalledWith(
"ws://127.0.0.1:25463/hub",
"",
);
expect(kill).toHaveBeenCalledWith(12345, "SIGTERM");
expect(clearHubDiscovery).toHaveBeenCalledWith("/tmp/hub-discovery.json");
expect(spawn).toHaveBeenCalledOnce();
} finally {
kill.mockRestore();
}
});
it("throws a targeted error when an incompatible hub cannot be retired", async () => {
vi.useFakeTimers();
try {
readHubDiscovery.mockResolvedValue(undefined);
probeHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v2",
buildId: "future-build",
});
const { ensureDetachedHubServer } = await import(".");
const pending = expect(
ensureDetachedHubServer("/workspace"),
).rejects.toThrow(
"An incompatible Cline Hub is already running at ws://127.0.0.1:25463/hub and could not be retired automatically.",
);
await vi.runAllTimersAsync();
await pending;
expect(spawn).not.toHaveBeenCalled();
} finally {
vi.useRealTimers();
}
});
it("retires a legacy shared production hub before resolving the production hub", async () => {
const kill = vi.spyOn(process, "kill").mockImplementation(() => true);
try {
resolveSharedHubOwnerContext.mockReturnValueOnce({
discoveryPath: "/tmp/legacy-hub-discovery.json",
});
readHubDiscovery
.mockResolvedValueOnce({
url: "ws://127.0.0.1:39121/hub",
authToken: "legacy-token",
pid: 222,
})
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
probeHubServer
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
});
verifyHubConnection.mockResolvedValueOnce(true);
const { ensureDetachedHubServer } = await import(".");
const result = await ensureDetachedHubServer("/workspace");
expect(result).toEqual({
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
expect(requestHubShutdown).toHaveBeenCalledWith(
"ws://127.0.0.1:39121/hub",
"legacy-token",
);
expect(kill).toHaveBeenCalledWith(222, "SIGTERM");
expect(clearHubDiscovery).toHaveBeenCalledWith(
"/tmp/legacy-hub-discovery.json",
);
expect(spawn).toHaveBeenCalledOnce();
} finally {
kill.mockRestore();
}
});
it("throws when a compatible expected hub has no discovery record", async () => {
readHubDiscovery.mockResolvedValue(undefined);
probeHubServer.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
});
const { ensureDetachedHubServer } = await import(".");
await expect(ensureDetachedHubServer("/workspace")).rejects.toThrow(
"A compatible Cline Hub is already running at ws://127.0.0.1:25463/hub, but its discovery record is missing or unreadable.",
);
expect(spawn).not.toHaveBeenCalled();
});
it("uses matching discovery pid and token when retiring an incompatible expected-url hub", async () => {
const kill = vi
.spyOn(process, "kill")
.mockImplementation((_pid, signal) => {
if (signal === 0) {
throw Object.assign(new Error("missing"), { code: "ESRCH" });
}
return true;
});
try {
readHubDiscovery
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
authToken: "old-token",
pid: 12345,
})
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
probeHubServer
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v2",
buildId: "future-build",
})
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
});
verifyHubConnection.mockResolvedValueOnce(true);
const { ensureDetachedHubServer } = await import(".");
const result = await ensureDetachedHubServer("/workspace");
expect(result).toEqual({
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
expect(requestHubShutdown).toHaveBeenCalledWith(
"ws://127.0.0.1:25463/hub",
"old-token",
);
expect(kill).toHaveBeenCalledWith(12345, "SIGTERM");
} finally {
kill.mockRestore();
}
});
it("does not reuse a healthy hub without protocol metadata", async () => {
const kill = vi.spyOn(process, "kill").mockImplementation(() => true);
try {
readHubDiscovery
@@ -311,7 +528,8 @@ describe("ensureDetachedHubServer", () => {
authToken: "old-token",
})
.mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
authToken: "new-token",
});
@@ -327,7 +545,13 @@ describe("ensureDetachedHubServer", () => {
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce(undefined)
.mockResolvedValueOnce({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
})
.mockResolvedValue({
url: "ws://127.0.0.1:25463/hub",
protocolVersion: "v1",
buildId: "current-build",
});
verifyHubConnection.mockResolvedValueOnce(true);
@@ -336,7 +560,7 @@ describe("ensureDetachedHubServer", () => {
const result = await ensureDetachedHubServer("/workspace");
expect(result).toEqual({
url: "ws://127.0.0.1:5555/hub",
url: "ws://127.0.0.1:25463/hub",
authToken: "new-token",
});
expect(requestHubShutdown).toHaveBeenCalledWith(
+207 -66
View File
@@ -5,6 +5,8 @@ import { fileURLToPath } from "node:url";
import {
CLINE_RUN_AS_HUB_DAEMON_ENV,
isHubDaemonProcess,
isHubProtocolCompatible,
resolveClineBuildEnv,
withResolvedClineBuildEnv,
} from "@cline/shared";
import {
@@ -15,17 +17,20 @@ import {
import {
clearHubDiscovery,
createHubServerUrl,
type HubServerDiscoveryRecord,
type HubOwnerContext,
type HubServerProbeRecord,
probeHubServer,
readHubDiscovery,
resolveClineDataDir,
resolveHubBuildId,
} from "../discovery";
import {
type HubEndpointOverrides,
resolveHubEndpointOptions,
} from "../discovery/defaults";
import { resolveSharedHubOwnerContext } from "../discovery/workspace";
import {
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
} from "../discovery/workspace";
const HUB_STARTUP_TIMEOUT_MS = 8_000;
const HUB_STARTUP_POLL_MS = 200;
@@ -54,16 +59,37 @@ function openDetachedHubLogFile(): { fd: number; logPath: string } | undefined {
}
}
function isCompatibleHubRecord(record: HubServerDiscoveryRecord): boolean {
const recordBuildId = record.buildId?.trim();
return !!recordBuildId && recordBuildId === resolveHubBuildId();
function resolveDefaultHubOwnerContext() {
return resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext();
}
function isCompatibleHubRecord(record: HubServerProbeRecord): boolean {
return isHubProtocolCompatible(record).compatible;
}
function withMatchingDiscoveryRetirementMetadata(
probe: HubServerProbeRecord,
discovered: { url?: string; authToken?: string; pid?: number } | undefined,
expectedUrl: string,
): HubServerProbeRecord {
if (!discovered || discovered.url !== expectedUrl) {
return probe;
}
return {
...probe,
authToken: probe.authToken ?? discovered.authToken,
pid: probe.pid ?? discovered.pid,
};
}
async function safeProbeHubServer(
url: string,
): Promise<HubServerDiscoveryRecord | undefined> {
authToken?: string,
): Promise<HubServerProbeRecord | undefined> {
try {
return await probeHubServer(url);
return await probeHubServer(url, { authToken });
} catch {
return undefined;
}
@@ -84,13 +110,10 @@ async function waitForHubToRetire(
return false;
}
async function retireIncompatibleHub(
record: HubServerDiscoveryRecord,
async function retireDiscoveredHub(
record: { url: string; authToken?: string; pid?: number },
discoveryPath: string,
): Promise<void> {
if (isCompatibleHubRecord(record)) {
return;
}
): Promise<boolean> {
await requestHubShutdown(record.url, record.authToken).catch(() => false);
if (record.pid) {
try {
@@ -99,8 +122,43 @@ async function retireIncompatibleHub(
// Best-effort cleanup only. A compatible hub may still start on a fallback port.
}
}
await waitForHubToRetire(record.url, HUB_RETIRE_TIMEOUT_MS);
const retired = await waitForHubToRetire(record.url, HUB_RETIRE_TIMEOUT_MS);
await clearHubDiscovery(discoveryPath).catch(() => undefined);
return retired;
}
async function retireIncompatibleHub(
record: HubServerProbeRecord,
discoveryPath: string,
): Promise<boolean> {
if (isCompatibleHubRecord(record)) {
return true;
}
return retireDiscoveredHub(record, discoveryPath);
}
/**
* Pre-singleton production builds tracked the local hub under the shared
* owner discovery path and spawned daemons on random fallback ports. Those
* daemons are invisible to the production owner context, so nothing would
* ever reuse or stop them. Retire the recorded legacy hub (its record carries
* the auth token and pid needed for a graceful stop) and clear the legacy
* record so upgrades do not leave orphaned daemons running stale code.
*/
async function retireLegacySharedHub(owner: HubOwnerContext): Promise<void> {
if (resolveClineBuildEnv() !== "production") {
return;
}
const legacy = resolveSharedHubOwnerContext();
if (legacy.discoveryPath === owner.discoveryPath) {
return;
}
const record = await readHubDiscovery(legacy.discoveryPath);
if (record?.url) {
await retireDiscoveredHub(record, legacy.discoveryPath);
} else {
await clearHubDiscovery(legacy.discoveryPath).catch(() => undefined);
}
}
function resolveDaemonEntryPath(): string {
@@ -200,48 +258,75 @@ export async function spawnDetachedHubServerWithRetry(
export function prewarmDetachedHubServer(
workspaceRoot: string,
endpoint: HubEndpointOverrides = {},
endpoint: HubEndpointOverrides & { allowPortFallback?: boolean } = {},
): void {
if (isHubDaemonProcess()) {
return;
}
const owner = resolveSharedHubOwnerContext();
const hasExplicitPort =
endpoint.port !== undefined || !!process.env.CLINE_HUB_PORT?.trim();
const owner = resolveDefaultHubOwnerContext();
const resolvedEndpoint = resolveHubEndpointOptions(endpoint);
const expectedUrl = createHubServerUrl(
resolvedEndpoint.host,
resolvedEndpoint.port,
resolvedEndpoint.pathname,
);
void readHubDiscovery(owner.discoveryPath)
const shouldUseFallbackPort =
endpoint.allowPortFallback === true && resolvedEndpoint.port !== 0;
void retireLegacySharedHub(owner)
.catch(() => undefined)
.then(() => readHubDiscovery(owner.discoveryPath))
.then(async (discovered) => {
let retiredUnusableDiscovery = false;
if (discovered?.url) {
const healthy = await safeProbeHubServer(discovered.url);
if (
healthy?.url &&
isCompatibleHubRecord(healthy) &&
(await verifyHubConnection(healthy.url, {
authToken: discovered.authToken,
}))
) {
return;
}
if (healthy?.url) {
await retireIncompatibleHub(
{ ...healthy, authToken: discovered.authToken },
if (!discovered.authToken) {
retiredUnusableDiscovery = true;
const retired = await retireDiscoveredHub(
discovered,
owner.discoveryPath,
);
if (!retired && !shouldUseFallbackPort) {
return;
}
} else {
await clearHubDiscovery(owner.discoveryPath).catch(() => undefined);
const healthy = await safeProbeHubServer(
discovered.url,
discovered.authToken,
);
if (
healthy?.url &&
isCompatibleHubRecord(healthy) &&
(await verifyHubConnection(healthy.url, {
authToken: discovered.authToken,
}))
) {
return;
}
if (healthy?.url) {
await retireIncompatibleHub(
{ ...healthy, authToken: discovered.authToken },
owner.discoveryPath,
);
} else {
await clearHubDiscovery(owner.discoveryPath).catch(() => undefined);
}
}
}
const expected = await safeProbeHubServer(expectedUrl);
if (expected?.url) {
await retireIncompatibleHub(expected, owner.discoveryPath);
if (isCompatibleHubRecord(expected)) {
if (!shouldUseFallbackPort || !retiredUnusableDiscovery) {
return;
}
} else {
const retiredExpected = await retireIncompatibleHub(
{ ...expected, authToken: undefined },
owner.discoveryPath,
);
if (!retiredExpected && !shouldUseFallbackPort) {
return;
}
}
}
const shouldUseFallbackPort =
!hasExplicitPort && resolvedEndpoint.port !== 0;
const spawnEndpoint = shouldUseFallbackPort
? { ...resolvedEndpoint, port: 0 }
: resolvedEndpoint;
@@ -259,17 +344,16 @@ export interface DetachedHubResolution {
export async function ensureDetachedHubServer(
workspaceRoot: string,
endpointOverrides: HubEndpointOverrides = {},
endpointOverrides: HubEndpointOverrides & {
allowPortFallback?: boolean;
} = {},
): Promise<DetachedHubResolution> {
const owner = resolveSharedHubOwnerContext();
const owner = resolveDefaultHubOwnerContext();
const hasExplicitEndpoint =
endpointOverrides.host !== undefined ||
endpointOverrides.port !== undefined ||
endpointOverrides.pathname !== undefined ||
!!process.env.CLINE_HUB_PORT?.trim();
const hasExplicitPort =
endpointOverrides.port !== undefined ||
!!process.env.CLINE_HUB_PORT?.trim();
const endpoint = resolveHubEndpointOptions(endpointOverrides);
const expectedUrl = createHubServerUrl(
endpoint.host,
@@ -284,35 +368,72 @@ export async function ensureDetachedHubServer(
}
return result;
};
await retireLegacySharedHub(owner).catch(() => undefined);
const discovered = await readHubDiscovery(owner.discoveryPath);
let retiredUnusableDiscovery = false;
if (discovered?.url) {
const healthy = await safeProbeHubServer(discovered.url);
if (
healthy?.url &&
isCompatibleHubRecord(healthy) &&
(await verifyHubConnection(healthy.url, {
authToken: discovered.authToken,
}))
) {
return rememberIfManaged({
url: healthy.url,
authToken: discovered.authToken,
});
}
if (healthy?.url) {
await retireIncompatibleHub(
{ ...healthy, authToken: discovered.authToken },
owner.discoveryPath,
);
const discoveredAuthToken = discovered.authToken;
if (!discoveredAuthToken) {
retiredUnusableDiscovery = true;
await retireDiscoveredHub(discovered, owner.discoveryPath);
} else {
await clearHubDiscovery(owner.discoveryPath).catch(() => undefined);
const healthy = await safeProbeHubServer(
discovered.url,
discoveredAuthToken,
);
if (
healthy?.url &&
isCompatibleHubRecord(healthy) &&
(await verifyHubConnection(healthy.url, {
authToken: discoveredAuthToken,
}))
) {
return rememberIfManaged({
url: healthy.url,
authToken: discoveredAuthToken,
});
}
if (healthy?.url) {
await retireIncompatibleHub(
{ ...healthy, authToken: discoveredAuthToken },
owner.discoveryPath,
);
} else {
await clearHubDiscovery(owner.discoveryPath).catch(() => undefined);
}
}
}
const expected = await safeProbeHubServer(expectedUrl);
if (expected?.url) {
await retireIncompatibleHub(expected, owner.discoveryPath);
const expectedForRetirement = withMatchingDiscoveryRetirementMetadata(
expected,
discovered,
expectedUrl,
);
if (isCompatibleHubRecord(expected)) {
const upgradeHint = retiredUnusableDiscovery
? " This can happen immediately after upgrading from a build that wrote an empty hub auth token; run 'cline doctor fix' to stop the old daemon and repair local hub discovery."
: "";
throw new Error(
`A compatible Cline Hub is already running at ${expectedUrl}, but its discovery record is missing or unreadable. Run 'cline doctor fix' to repair local hub discovery.${upgradeHint}`,
);
}
const retiredExpected = await retireIncompatibleHub(
expectedForRetirement,
owner.discoveryPath,
);
if (
!retiredExpected &&
endpointOverrides.allowPortFallback !== true &&
endpoint.port !== 0
) {
throw new Error(
`An incompatible Cline Hub is already running at ${expectedUrl} and could not be retired automatically. Run 'cline doctor fix' to stop stale hub daemons before starting a new hub.`,
);
}
}
const shouldUseFallbackPort = !hasExplicitPort && endpoint.port !== 0;
const shouldUseFallbackPort =
endpointOverrides.allowPortFallback === true && endpoint.port !== 0;
const spawnEndpoint = shouldUseFallbackPort
? { ...endpoint, port: 0 }
: endpoint;
@@ -320,8 +441,11 @@ export async function ensureDetachedHubServer(
const deadline = Date.now() + HUB_STARTUP_TIMEOUT_MS;
while (Date.now() < deadline) {
const nextDiscovery = await readHubDiscovery(owner.discoveryPath);
if (nextDiscovery?.url) {
const healthy = await safeProbeHubServer(nextDiscovery.url);
if (nextDiscovery?.url && nextDiscovery.authToken) {
const healthy = await safeProbeHubServer(
nextDiscovery.url,
nextDiscovery.authToken,
);
if (
healthy?.url &&
isCompatibleHubRecord(healthy) &&
@@ -337,7 +461,24 @@ export async function ensureDetachedHubServer(
}
const nextExpected = await safeProbeHubServer(expectedUrl);
if (nextExpected?.url && !isCompatibleHubRecord(nextExpected)) {
await retireIncompatibleHub(nextExpected, owner.discoveryPath);
const expectedForRetirement = withMatchingDiscoveryRetirementMetadata(
nextExpected,
nextDiscovery,
expectedUrl,
);
const retiredExpected = await retireIncompatibleHub(
expectedForRetirement,
owner.discoveryPath,
);
if (
!retiredExpected &&
endpointOverrides.allowPortFallback !== true &&
endpoint.port !== 0
) {
throw new Error(
`An incompatible Cline Hub is still running at ${expectedUrl} and could not be retired automatically. Run 'cline doctor fix' to stop stale hub daemons before starting a new hub.`,
);
}
}
await new Promise((resolve) => setTimeout(resolve, HUB_STARTUP_POLL_MS));
}
@@ -0,0 +1,135 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import type { EnsureHubServerOptions } from "./start-shared-server";
const {
mockEnsureHubWebSocketServer,
mockResolveHubEndpointOptions,
mockResolveClineBuildEnv,
mockResolveProductionHubOwnerContext,
mockResolveSharedHubOwnerContext,
mockStartHubWebSocketServer,
} = vi.hoisted(() => ({
mockEnsureHubWebSocketServer: vi.fn(async () => ({
url: "ws://127.0.0.1:25463/hub",
authToken: "token",
action: "started",
})),
mockResolveHubEndpointOptions: vi.fn(
(options: { host?: string; port?: number; pathname?: string }) => ({
host: options.host ?? "127.0.0.1",
port: options.port ?? 25463,
pathname: options.pathname ?? "/hub",
}),
),
mockResolveClineBuildEnv: vi.fn(() => "production"),
mockResolveProductionHubOwnerContext: vi.fn(() => ({
ownerId: "production",
discoveryPath: "/tmp/cline-data/locks/hub/production.json",
})),
mockResolveSharedHubOwnerContext: vi.fn(() => ({
ownerId: "shared",
discoveryPath: "/tmp/cline-data/locks/hub/owners/shared.json",
})),
mockStartHubWebSocketServer: vi.fn(),
}));
vi.mock("@cline/shared", () => ({
resolveClineBuildEnv: mockResolveClineBuildEnv,
}));
vi.mock("../discovery/defaults", () => ({
resolveHubEndpointOptions: mockResolveHubEndpointOptions,
}));
vi.mock("../discovery/workspace", () => ({
resolveProductionHubOwnerContext: mockResolveProductionHubOwnerContext,
resolveSharedHubOwnerContext: mockResolveSharedHubOwnerContext,
}));
vi.mock("../server", () => ({
ensureHubWebSocketServer: mockEnsureHubWebSocketServer,
startHubWebSocketServer: mockStartHubWebSocketServer,
}));
const originalHubPort = process.env.CLINE_HUB_PORT;
const runtimeHandlers =
{} as unknown as EnsureHubServerOptions["runtimeHandlers"];
describe("ensureHubServer", () => {
afterEach(() => {
mockEnsureHubWebSocketServer.mockClear();
mockResolveHubEndpointOptions.mockClear();
mockResolveClineBuildEnv.mockClear();
mockResolveClineBuildEnv.mockReturnValue("production");
mockResolveProductionHubOwnerContext.mockClear();
mockResolveSharedHubOwnerContext.mockClear();
mockStartHubWebSocketServer.mockClear();
if (originalHubPort === undefined) {
delete process.env.CLINE_HUB_PORT;
} else {
process.env.CLINE_HUB_PORT = originalHubPort;
}
});
it("does not allow port fallback by default in production", async () => {
delete process.env.CLINE_HUB_PORT;
const { ensureHubServer } = await import("./start-shared-server");
await ensureHubServer({ runtimeHandlers });
expect(mockEnsureHubWebSocketServer).toHaveBeenCalledWith(
expect.objectContaining({
port: 25463,
allowPortFallback: false,
owner: expect.objectContaining({
discoveryPath: "/tmp/cline-data/locks/hub/production.json",
}),
}),
);
});
it("allows port fallback by default in development when no port is explicit", async () => {
delete process.env.CLINE_HUB_PORT;
mockResolveClineBuildEnv.mockReturnValue("development");
const { ensureHubServer } = await import("./start-shared-server");
await ensureHubServer({ runtimeHandlers });
expect(mockEnsureHubWebSocketServer).toHaveBeenCalledWith(
expect.objectContaining({
port: 25463,
allowPortFallback: true,
owner: expect.objectContaining({
discoveryPath: "/tmp/cline-data/locks/hub/owners/shared.json",
}),
}),
);
});
it("does not default port fallback when a port option is explicit", async () => {
delete process.env.CLINE_HUB_PORT;
const { ensureHubServer } = await import("./start-shared-server");
await ensureHubServer({ port: 30000, runtimeHandlers });
expect(mockEnsureHubWebSocketServer).toHaveBeenCalledWith(
expect.objectContaining({
port: 30000,
allowPortFallback: false,
}),
);
});
it("does not default port fallback when CLINE_HUB_PORT is explicit", async () => {
process.env.CLINE_HUB_PORT = "30001";
const { ensureHubServer } = await import("./start-shared-server");
await ensureHubServer({ runtimeHandlers });
expect(mockEnsureHubWebSocketServer).toHaveBeenCalledWith(
expect.objectContaining({
allowPortFallback: false,
}),
);
});
});
@@ -1,5 +1,9 @@
import { resolveClineBuildEnv } from "@cline/shared";
import { resolveHubEndpointOptions } from "../discovery/defaults";
import { resolveSharedHubOwnerContext } from "../discovery/workspace";
import {
resolveProductionHubOwnerContext,
resolveSharedHubOwnerContext,
} from "../discovery/workspace";
import {
type EnsuredHubWebSocketServerResult,
type EnsureHubWebSocketServerOptions,
@@ -18,9 +22,19 @@ export interface StartHubServerOptions
export interface EnsureHubServerOptions
extends Omit<EnsureHubWebSocketServerOptions, "owner"> {}
function resolveDefaultHubOwnerContext() {
return resolveClineBuildEnv() === "production"
? resolveProductionHubOwnerContext()
: resolveSharedHubOwnerContext();
}
function shouldAllowDefaultPortFallback(hasExplicitPort: boolean): boolean {
return resolveClineBuildEnv() !== "production" && !hasExplicitPort;
}
/**
* Start a hub WebSocket server bound to the process-local shared owner
* context. Callers that need a custom owner should invoke
* Start a hub WebSocket server bound to the default owner context for the
* current build environment. Callers that need a custom owner should invoke
* {@link startHubWebSocketServer} directly.
*/
export async function startHubServer(
@@ -34,13 +48,13 @@ export async function startHubServer(
return await startHubWebSocketServer({
...options,
...endpoint,
owner: resolveSharedHubOwnerContext(),
owner: resolveDefaultHubOwnerContext(),
});
}
/**
* Ensure a hub WebSocket server is running in the process-local shared owner
* context, reusing a compatible in-process instance when available.
* Ensure a hub WebSocket server is running in the default owner context for the
* current build environment, reusing a compatible in-process instance when available.
*/
export async function ensureHubServer(
options: EnsureHubServerOptions,
@@ -55,7 +69,9 @@ export async function ensureHubServer(
return await ensureHubWebSocketServer({
...options,
...endpoint,
allowPortFallback: options.allowPortFallback ?? !hasExplicitPort,
owner: resolveSharedHubOwnerContext(),
allowPortFallback:
options.allowPortFallback ??
shouldAllowDefaultPortFallback(hasExplicitPort),
owner: resolveDefaultHubOwnerContext(),
});
}
@@ -3,6 +3,7 @@ import { dirname, join } from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import {
clearHubDiscovery,
probeHubServer,
readHubDiscovery,
resolveHubOwnerContext,
writeHubDiscovery,
@@ -88,4 +89,62 @@ describe("hub discovery", () => {
await clearHubDiscovery(discoveryPath);
await expect(readHubDiscovery(discoveryPath)).resolves.toBeUndefined();
});
it("rejects discovery records without an auth token", async () => {
snapshot = captureEnv();
delete process.env.CLINE_HUB_DISCOVERY_PATH;
process.env.CLINE_DATA_DIR = "/tmp/cline-data";
const discoveryPath = resolveHubOwnerContext("missing-auth").discoveryPath;
await mkdir(dirname(discoveryPath), { recursive: true });
await writeFile(
discoveryPath,
`${JSON.stringify({
hubId: "hub_123",
protocolVersion: "v1",
host: "127.0.0.1",
port: 25463,
url: "ws://127.0.0.1:25463/hub",
startedAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
})}\n`,
"utf8",
);
await expect(readHubDiscovery(discoveryPath)).resolves.toBeUndefined();
});
it("returns only public health fields for unauthenticated probes", async () => {
const fetchMock = async () =>
({
ok: true,
json: async () => ({
ok: true,
protocolVersion: "v1",
minClientProtocolVersion: "v1",
maxClientProtocolVersion: "v1",
coreVersion: "1.0.0",
host: "127.0.0.1",
port: 25463,
url: "ws://127.0.0.1:25463/hub",
}),
}) as Response;
const originalFetch = globalThis.fetch;
globalThis.fetch = fetchMock as typeof fetch;
try {
const record = await probeHubServer("ws://127.0.0.1:25463/hub");
expect(record).toMatchObject({
protocolVersion: "v1",
host: "127.0.0.1",
port: 25463,
url: "ws://127.0.0.1:25463/hub",
});
expect(record?.hubId).toBeUndefined();
expect(record?.startedAt).toBeUndefined();
expect(record?.updatedAt).toBeUndefined();
} finally {
globalThis.fetch = originalFetch;
}
});
});
+90 -3
View File
@@ -14,6 +14,9 @@ const HUB_STARTUP_LOCK_POLL_MS = 100;
export interface HubServerDiscoveryRecord {
hubId: string;
protocolVersion: string;
minClientProtocolVersion?: string;
maxClientProtocolVersion?: string;
capabilities?: readonly string[];
coreVersion?: string;
buildId?: string;
authToken: string;
@@ -25,6 +28,23 @@ export interface HubServerDiscoveryRecord {
updatedAt: string;
}
export type HubServerProbeRecord = {
protocolVersion: string;
minClientProtocolVersion?: string;
maxClientProtocolVersion?: string;
capabilities?: readonly string[];
coreVersion?: string;
buildId?: string;
host: string;
port: number;
url: string;
hubId?: string;
authToken?: string;
pid?: number;
startedAt?: string;
updatedAt?: string;
};
export interface HubOwnerContext {
ownerId: string;
discoveryPath: string;
@@ -135,6 +155,20 @@ export async function readHubDiscovery(
return {
hubId: parsed.hubId,
protocolVersion: parsed.protocolVersion,
minClientProtocolVersion:
typeof parsed.minClientProtocolVersion === "string"
? parsed.minClientProtocolVersion
: undefined,
maxClientProtocolVersion:
typeof parsed.maxClientProtocolVersion === "string"
? parsed.maxClientProtocolVersion
: undefined,
capabilities: Array.isArray(parsed.capabilities)
? parsed.capabilities.filter(
(capability): capability is string =>
typeof capability === "string",
)
: undefined,
coreVersion:
typeof parsed.coreVersion === "string" ? parsed.coreVersion : undefined,
buildId: typeof parsed.buildId === "string" ? parsed.buildId : undefined,
@@ -225,13 +259,60 @@ export async function withHubStartupLock<T>(
export async function probeHubServer(
url: string,
): Promise<HubServerDiscoveryRecord | undefined> {
options?: { authToken?: string },
): Promise<HubServerProbeRecord | undefined> {
try {
const response = await fetch(toHubHealthUrl(url));
const response = await fetch(
options?.authToken ? toHubStatusUrl(url) : toHubHealthUrl(url),
{
headers: options?.authToken
? { authorization: `Bearer ${options.authToken}` }
: undefined,
},
);
if (!response.ok) {
return undefined;
}
return (await response.json()) as HubServerDiscoveryRecord;
const parsed = (await response.json()) as Partial<HubServerProbeRecord>;
if (
typeof parsed.protocolVersion !== "string" ||
typeof parsed.host !== "string" ||
typeof parsed.port !== "number" ||
typeof parsed.url !== "string"
) {
return undefined;
}
return {
protocolVersion: parsed.protocolVersion,
minClientProtocolVersion:
typeof parsed.minClientProtocolVersion === "string"
? parsed.minClientProtocolVersion
: undefined,
maxClientProtocolVersion:
typeof parsed.maxClientProtocolVersion === "string"
? parsed.maxClientProtocolVersion
: undefined,
capabilities: Array.isArray(parsed.capabilities)
? parsed.capabilities.filter(
(capability): capability is string =>
typeof capability === "string",
)
: undefined,
coreVersion:
typeof parsed.coreVersion === "string" ? parsed.coreVersion : undefined,
buildId: typeof parsed.buildId === "string" ? parsed.buildId : undefined,
host: parsed.host,
port: parsed.port,
url: parsed.url,
hubId: typeof parsed.hubId === "string" ? parsed.hubId : undefined,
authToken:
typeof parsed.authToken === "string" ? parsed.authToken : undefined,
pid: typeof parsed.pid === "number" ? parsed.pid : undefined,
startedAt:
typeof parsed.startedAt === "string" ? parsed.startedAt : undefined,
updatedAt:
typeof parsed.updatedAt === "string" ? parsed.updatedAt : undefined,
};
} catch {
return undefined;
}
@@ -253,6 +334,12 @@ export function toHubHealthUrl(wsUrl: string): string {
return parsed.toString();
}
export function toHubStatusUrl(wsUrl: string): string {
const parsed = new URL(toHubHealthUrl(wsUrl));
parsed.pathname = "/status";
return parsed.toString();
}
export function isDiscoveryFilePresent(pathname: string): boolean {
return existsSync(pathname);
}
@@ -1,7 +1,14 @@
import { join } from "node:path";
import { normalizeWorkspacePath } from "../../services/workspace/workspace-manifest";
import { type HubOwnerContext, resolveHubOwnerContext } from ".";
import {
type HubOwnerContext,
resolveClineDataDir,
resolveHubOwnerContext,
} from ".";
const DEFAULT_SHARED_HUB_OWNER_LABEL = "shared:cline";
const HUB_DISCOVERY_ENV = "CLINE_HUB_DISCOVERY_PATH";
const PRODUCTION_HUB_OWNER_ID = "hub-production";
export function resolveWorkspaceHubOwnerContext(
workspaceRoot: string,
@@ -17,3 +24,12 @@ export function resolveSharedHubOwnerContext(
): HubOwnerContext {
return resolveHubOwnerContext(label);
}
export function resolveProductionHubOwnerContext(): HubOwnerContext {
return {
ownerId: PRODUCTION_HUB_OWNER_ID,
discoveryPath:
process.env[HUB_DISCOVERY_ENV]?.trim() ||
join(resolveClineDataDir(), "locks", "hub", "production.json"),
};
}
@@ -0,0 +1,21 @@
import { describe, expect, it } from "vitest";
import { readBearerToken } from "./hub-websocket-server";
describe("readBearerToken", () => {
it("reads a bearer token with case-insensitive scheme", () => {
expect(readBearerToken("Bearer token")).toBe("token");
expect(readBearerToken("bearer token")).toBe("token");
});
it("reads a bearer token separated by tabs without regex backtracking", () => {
expect(readBearerToken(`bearer\t\t${"token"}`)).toBe("token");
expect(readBearerToken(`bearer${"\t".repeat(10_000)}token`)).toBe("token");
});
it("rejects missing and malformed bearer tokens", () => {
expect(readBearerToken(undefined)).toBeNull();
expect(readBearerToken("Bearer")).toBeNull();
expect(readBearerToken("BearerToken")).toBeNull();
expect(readBearerToken("Basic token")).toBeNull();
});
});
@@ -2,6 +2,13 @@ import { timingSafeEqual } from "node:crypto";
import http from "node:http";
import net from "node:net";
import { URL } from "node:url";
import {
CURRENT_HUB_PROTOCOL_VERSION,
HUB_CAPABILITIES,
isHubProtocolCompatible,
MAX_CLIENT_HUB_PROTOCOL_VERSION,
MIN_CLIENT_HUB_PROTOCOL_VERSION,
} from "@cline/shared";
import { WebSocketServer } from "ws";
import corePackage from "../../../package.json";
import { rememberRecoverableLocalHubUrl, verifyHubConnection } from "../client";
@@ -204,10 +211,32 @@ function parseHeaderValue(value: string | string[] | undefined): string {
return Array.isArray(value) ? value.join(",") : (value ?? "");
}
function readBearerToken(value: string | string[] | undefined): string | null {
function isAuthHeaderWhitespace(code: number): boolean {
return code === 0x20 || code === 0x09;
}
export function readBearerToken(
value: string | string[] | undefined,
): string | null {
const header = parseHeaderValue(value).trim();
const match = /^Bearer\s+(.+)$/i.exec(header);
return match?.[1]?.trim() || null;
const bearerScheme = "bearer";
if (
header.length <= bearerScheme.length ||
header.slice(0, bearerScheme.length).toLowerCase() !== bearerScheme ||
!isAuthHeaderWhitespace(header.charCodeAt(bearerScheme.length))
) {
return null;
}
let tokenStart = bearerScheme.length + 1;
while (
tokenStart < header.length &&
isAuthHeaderWhitespace(header.charCodeAt(tokenStart))
) {
tokenStart += 1;
}
return header.slice(tokenStart).trim() || null;
}
function readWebSocketAuthToken(
@@ -244,7 +273,10 @@ export async function startHubWebSocketServer(
const cleanup = new Set<() => void>();
const startedAt = new Date().toISOString();
const versionPayload = {
protocolVersion: "v1",
protocolVersion: CURRENT_HUB_PROTOCOL_VERSION,
minClientProtocolVersion: MIN_CLIENT_HUB_PROTOCOL_VERSION,
maxClientProtocolVersion: MAX_CLIENT_HUB_PROTOCOL_VERSION,
capabilities: HUB_CAPABILITIES,
coreVersion: corePackage.version,
buildId,
pid: process.pid,
@@ -300,10 +332,36 @@ export async function startHubWebSocketServer(
const server = http.createServer((req, res) => {
if ((req.url ?? "/") === "/health") {
const body = JSON.stringify({
ok: true,
protocolVersion: versionPayload.protocolVersion,
minClientProtocolVersion: versionPayload.minClientProtocolVersion,
maxClientProtocolVersion: versionPayload.maxClientProtocolVersion,
coreVersion: versionPayload.coreVersion,
host,
port,
url,
});
res.statusCode = 200;
res.setHeader("content-type", "application/json");
res.end(body);
return;
}
if ((req.url ?? "/") === "/status") {
if (
!isValidHubAuthToken(
readBearerToken(req.headers.authorization),
authToken,
)
) {
res.statusCode = 401;
res.end("Unauthorized");
return;
}
const body = JSON.stringify({
hubId: transport.getHubId(),
...versionPayload,
authToken: "",
authToken,
host,
port,
url,
@@ -449,7 +507,10 @@ export async function startHubWebSocketServer(
await writeHubDiscovery(owner.discoveryPath, {
hubId: transport.getHubId(),
protocolVersion: "v1",
protocolVersion: CURRENT_HUB_PROTOCOL_VERSION,
minClientProtocolVersion: MIN_CLIENT_HUB_PROTOCOL_VERSION,
maxClientProtocolVersion: MAX_CLIENT_HUB_PROTOCOL_VERSION,
capabilities: [...versionPayload.capabilities],
coreVersion: corePackage.version,
buildId,
authToken,
@@ -511,9 +572,12 @@ export async function ensureHubWebSocketServer(
discovered?.url &&
(discovered.url === expectedUrl || options.allowPortFallback === true);
if (canReuseDiscovered) {
const healthy = await probeHubServer(discovered.url);
const healthy = await probeHubServer(discovered.url, {
authToken: discovered.authToken,
});
if (
healthy?.url &&
isHubProtocolCompatible(healthy).compatible &&
(await verifyHubConnection(healthy.url, {
authToken: discovered.authToken,
}))
@@ -526,8 +590,9 @@ export async function ensureHubWebSocketServer(
}
}
const expected = await probeHubServer(expectedUrl);
if (expected?.url || discovered?.url) {
// The discovered hub was not reusable (missing, mismatched, or failed
// verification), so its record is stale either way.
if (discovered?.url) {
await clearHubDiscovery(owner.discoveryPath);
}
+31
View File
@@ -0,0 +1,31 @@
import { describe, expect, it } from "vitest";
import { isHubProtocolCompatible } from "./hub";
describe("isHubProtocolCompatible", () => {
it("accepts a hub whose supported client range includes the client protocol", () => {
expect(
isHubProtocolCompatible({
protocolVersion: "v2",
minClientProtocolVersion: "v1",
maxClientProtocolVersion: "v2",
}),
).toEqual({ compatible: true });
});
it("rejects a hub whose supported client range excludes the client protocol", () => {
expect(
isHubProtocolCompatible({
protocolVersion: "v2",
minClientProtocolVersion: "v2",
maxClientProtocolVersion: "v3",
}),
).toEqual({ compatible: false, reason: "unsupported_protocol" });
});
it("rejects missing or malformed protocol versions", () => {
expect(isHubProtocolCompatible({ protocolVersion: "" })).toEqual({
compatible: false,
reason: "missing_protocol",
});
});
});
+71
View File
@@ -5,6 +5,77 @@ import type { RuntimeConfigExtensionKind } from "./session/runtime-config";
export type HubProtocolVersion = "v1";
export const CURRENT_HUB_PROTOCOL_VERSION: HubProtocolVersion = "v1";
export const MIN_CLIENT_HUB_PROTOCOL_VERSION: HubProtocolVersion = "v1";
export const MAX_CLIENT_HUB_PROTOCOL_VERSION: HubProtocolVersion = "v1";
export type HubCapabilityName =
| "client.register"
| "client.list"
| "session.create"
| "session.list"
| "session.get"
| "session.run"
| "session.abort"
| "schedule.create"
| "schedule.list"
| "settings.get"
| "settings.set";
export const HUB_CAPABILITIES: readonly HubCapabilityName[] = [
"client.register",
"client.list",
"session.create",
"session.list",
"session.get",
"session.run",
"session.abort",
"schedule.create",
"schedule.list",
"settings.get",
"settings.set",
];
export interface HubProtocolMetadata {
protocolVersion: string;
minClientProtocolVersion?: string;
maxClientProtocolVersion?: string;
capabilities?: readonly string[];
}
export type HubCompatibilityResult =
| { compatible: true }
| { compatible: false; reason: "missing_protocol" | "unsupported_protocol" };
function parseHubProtocolNumber(
version: string | undefined,
): number | undefined {
const match = /^v(\d+)$/.exec(version?.trim() ?? "");
if (!match) {
return undefined;
}
return Number.parseInt(match[1] ?? "", 10);
}
export function isHubProtocolCompatible(
hub: HubProtocolMetadata,
clientProtocolVersion: HubProtocolVersion = CURRENT_HUB_PROTOCOL_VERSION,
): HubCompatibilityResult {
const hubProtocol = parseHubProtocolNumber(hub.protocolVersion);
const clientProtocol = parseHubProtocolNumber(clientProtocolVersion);
if (hubProtocol === undefined || clientProtocol === undefined) {
return { compatible: false, reason: "missing_protocol" };
}
const minClientProtocol =
parseHubProtocolNumber(hub.minClientProtocolVersion) ?? hubProtocol;
const maxClientProtocol =
parseHubProtocolNumber(hub.maxClientProtocolVersion) ?? hubProtocol;
return clientProtocol >= minClientProtocol &&
clientProtocol <= maxClientProtocol
? { compatible: true }
: { compatible: false, reason: "unsupported_protocol" };
}
export type HubActorKind = "client" | "peerHub";
export type HubTransportKind =