Stabilize scoped routing and reconnect verification

Author: gitcommit90

src/server/index.ts

@@ -1816,7 +1816,10 @@ async function bootstrap(): Promise<void> {
   registerWorkflowDispatcher((bot, channelId, triggerId, threadRootId) => runBot(bot, channelId, triggerId, threadRootId, true));
   reactivateComputersAfterPreparedRemoval();
   prepareMnemosyneRuntime();
-  await startRoutingEngine((activity) => broadcastAdmins({ type: "routing_activity", activity }));
+  await startRoutingEngine((activity, ownerUserId) => {
+    if (ownerUserId) sendToUsers([ownerUserId], { type: "routing_activity", activity });
+    else broadcastAdmins({ type: "routing_activity", activity });
+  });
   ensureImageGenerationSkill();
   await internalRoutingProviderId();
   resumeQueuedAgentTurns();

src/server/routing.ts

@@ -118,6 +118,8 @@ type UserGatewayRuntime = {
   port: number;
   gateway: UserGateway;
   router: RoutingRuntime["router"];
+  requestActivity: RoutingRuntime["requestActivity"];
+  activityUnsubscribe: () => void;
 };
 
 // ReRouted's generic OpenAI-compatible adapter historically assumed every
@@ -151,8 +153,9 @@ const INTERNAL_GATEWAY_KEY_SCOPE = "1helm-internal";
 let runtime: RoutingRuntime | null = null;
 let starting: Promise<RoutingRuntime> | null = null;
 let activityUnsubscribe: (() => void) | null = null;
-let onActivity: ((activity?: unknown) => void) | null = null;
+let onActivity: ((activity?: unknown, userId?: number) => void) | null = null;
 const recentActivity: unknown[] = [];
+const recentUserActivity = new Map<number, unknown[]>();
 type OauthCompletion = { connected: boolean; account?: Record<string, unknown>; error?: string };
 const oauthWatchers = new Map<string, NodeJS.Timeout>();
 const oauthCompletions = new Map<string, OauthCompletion>();
@@ -161,6 +164,15 @@ const oauthOwners = new Map<string, { userId: number; type: string; providerIds:
 const activeOauthByFamily = new Map<string, string>();
 const userGateways = new Map<number, UserGatewayRuntime>();
 
+function publishRoutingActivity(activity: unknown, userId = 0): void {
+  const history = userId > 0
+    ? (recentUserActivity.get(userId) || (recentUserActivity.set(userId, []), recentUserActivity.get(userId)!))
+    : recentActivity;
+  history.unshift(activity);
+  if (history.length > 30) history.length = 30;
+  onActivity?.(activity, userId);
+}
+
 const oauthKey = (userId: number, type: string): string => `${userId}:${type}`;
 const oauthFamily = (type: string): string => ["chatgpt", "codex"].includes(type) ? "chatgpt" : type;
 
@@ -276,10 +288,17 @@ async function ensureUserGateway(userId: number): Promise<UserGatewayRuntime> {
     },
   } as RoutingRuntime["router"];
   const gateway = createGateway({ store, router, requestActivity });
+  const activityUnsubscribe = requestActivity.subscribe((activity) => publishRoutingActivity(activity, userId));
   const desiredPort = await choosePort(endpoint.port, "127.0.0.1", false);
-  const address = await gateway.start(desiredPort, "127.0.0.1");
+  let address: { port: number; host: string };
+  try {
+    address = await gateway.start(desiredPort, "127.0.0.1");
+  } catch (error) {
+    activityUnsubscribe();
+    throw error;
+  }
   if (address.port !== endpoint.port) run("UPDATE user_routing_endpoints SET port=?,updated=? WHERE user_id=?", address.port, now(), userId);
-  const created = { userId, port: address.port, gateway, router };
+  const created = { userId, port: address.port, gateway, router, requestActivity, activityUnsubscribe };
   userGateways.set(userId, created);
   return created;
 }
@@ -701,7 +720,7 @@ async function choosePort(preferred: number, host: string, honorConfiguredPort =
   });
 }
 
-export async function startRoutingEngine(activityCallback?: (activity?: unknown) => void): Promise<RoutingRuntime> {
+export async function startRoutingEngine(activityCallback?: (activity?: unknown, userId?: number) => void): Promise<RoutingRuntime> {
   if (runtime) return runtime;
   if (starting) return starting;
   onActivity = activityCallback || null;
@@ -723,11 +742,7 @@ export async function startRoutingEngine(activityCallback?: (activity?: unknown)
     if (port !== Number(config.port || 4949)) target.store.update((current) => { current.port = port; });
     await target.start({ port, host });
     ensureInternalProvider(target);
-    activityUnsubscribe = target.requestActivity.subscribe((activity) => {
-      recentActivity.unshift(activity);
-      if (recentActivity.length > 30) recentActivity.length = 30;
-      onActivity?.(activity);
-    });
+    activityUnsubscribe = target.requestActivity.subscribe((activity) => publishRoutingActivity(activity));
     runtime = target;
     return target;
   })().finally(() => { starting = null; });
@@ -746,7 +761,10 @@ export async function stopRoutingEngine(): Promise<void> {
   activeOauthByFamily.clear();
   const gateways = [...userGateways.values()];
   userGateways.clear();
+  for (const entry of gateways) entry.activityUnsubscribe();
   await Promise.all(gateways.map((entry) => entry.gateway.stop().catch(() => undefined)));
+  recentActivity.length = 0;
+  recentUserActivity.clear();
   if (target) await target.close({ drainMs: 10_000 });
 }
 
@@ -955,7 +973,9 @@ export async function routingState(userId = 0, isAdmin = true): Promise<Record<s
     visibility: visibilityOf(combo),
     mine: Number(combo.ownerUserId || 0) === userId,
   }));
-  return { ...state, providers, combos, activeRequests: runtime?.requestActivity.snapshot() || [], recentActivity, imageGenerationEnabled: enabled, apiKey: state.apiKey ? "" : state.apiKey, apiKeys: undefined, scope: isAdmin ? "captain" : "member" };
+  const personalActivity = userId ? recentUserActivity.get(userId) || [] : recentActivity;
+  const activeRequests = userId ? userGateways.get(userId)?.requestActivity.snapshot() || [] : runtime?.requestActivity.snapshot() || [];
+  return { ...state, providers, combos, activeRequests, recentActivity: personalActivity, imageGenerationEnabled: enabled, apiKey: state.apiKey ? "" : state.apiKey, apiKeys: undefined, scope: isAdmin ? "captain" : "member" };
 }
 
 export async function routingCredentials(userId = 0, isAdmin = true): Promise<Record<string, unknown>> {

test/connectors.mjs

@@ -35,9 +35,15 @@ test("stopping a connector cancels automatic relaunch while preserving its crede
   const beforeStop = (await readFile(log, "utf8")).trim().split("\n").length;
   assert.ok(beforeStop >= 2, "a failed desired connector relaunches");
   connectors.stopTunnelConnector("workspace");
+  // A process already returned by spawn() can reach its first instruction
+  // while stop is delivering SIGTERM. Establish the stopped boundary after
+  // that in-flight child settles, then prove no timer can launch another one.
+  await new Promise((resolve) => setTimeout(resolve, 75));
+  const stoppedBoundary = (await readFile(log, "utf8")).trim().split("\n").length;
   await new Promise((resolve) => setTimeout(resolve, 140));
   const afterStop = (await readFile(log, "utf8")).trim().split("\n").length;
-  assert.equal(afterStop, beforeStop, "disabled connector does not relaunch");
+  assert.ok(stoppedBoundary >= beforeStop, "stop accounts for any child that was already spawned");
+  assert.equal(afterStop, stoppedBoundary, "disabled connector does not relaunch after the stopped boundary");
   assert.ok(connectors.connectorCredential("workspace"), "disabling leaves the reserved connector credential available for re-enable");
 
   process.env.HELM_CONNECTOR_TEST_HOLD = "1";

test/routing.mjs

@@ -162,6 +162,14 @@ test("embedded provider fabric powers 1Helm agents and its public endpoint", { t
     assert.notEqual(guestCredentials.personalPort, captainCredentialsBefore.personalPort, "each signed-in member receives a dedicated host-side endpoint port");
     assert.equal((await json(`http://127.0.0.1:${appPort}/v1/models`, guestExternal.key.key)).data.some((model) => model.id.includes("mock-large")), true, "member key routes through personal plus shared providers");
     await json(`http://127.0.0.1:${appPort}/api/routing/action`, guestToken, { method: "POST", body: JSON.stringify({ action: "app:set-provider-visibility", payload: { id: guestKeyed.id, visibility: "personal" } }) });
+    const guestPrivateModel = (await json(`http://127.0.0.1:${appPort}/api/routing/models`, guestToken)).models.find((model) => model.providerName === "Crew private").id;
+    await json(`http://127.0.0.1:${appPort}/v1/chat/completions`, guestExternal.key.key, {
+      method: "POST", body: JSON.stringify({ model: guestPrivateModel, stream: false, messages: [{ role: "user", content: "Private member activity" }] }),
+    });
+    const guestActivity = (await json(`http://127.0.0.1:${appPort}/api/routing/state`, guestToken)).recentActivity;
+    const captainActivity = (await json(`http://127.0.0.1:${appPort}/api/routing/state`, token)).recentActivity;
+    assert.equal(guestActivity.some((event) => event.request?.providerName === "Crew private"), true, "personal endpoint activity is visible to its owner");
+    assert.equal(captainActivity.some((event) => event.request?.providerName === "Crew private"), false, "personal endpoint activity is not disclosed to the Captain or another member");
     assert.equal((await json(`http://127.0.0.1:${appPort}/api/routing/state`, token)).combos.some((entry) => entry.name === "crew-shared-route"), false, "a shared route stops resolving for teammates when its provider becomes private");
     const guestOwnedSharedRoute = (await json(`http://127.0.0.1:${appPort}/api/routing/state`, guestToken)).combos.find((entry) => entry.name === "crew-shared-route");
     await json(`http://127.0.0.1:${appPort}/api/workspace/model-policy`, guestToken, { method: "PATCH", body: JSON.stringify({ model: "crew-shared-route", personal: true }) });

test/terminal-reconnect-browser.mjs

@@ -1,13 +1,30 @@
 import assert from "node:assert/strict";
 import { spawn } from "node:child_process";
 import { createServer } from "node:net";
-import { mkdtempSync, rmSync } from "node:fs";
+import { accessSync, constants as fsConstants, mkdtempSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import test from "node:test";
 import puppeteer from "puppeteer";
 
 const root = new URL("..", import.meta.url).pathname;
+function browserExecutable() {
+  const configured = process.env.PUPPETEER_EXECUTABLE_PATH;
+  if (configured) {
+    try { accessSync(configured, fsConstants.X_OK); return configured; } catch { /* use discovery */ }
+  }
+  try {
+    const bundled = puppeteer.executablePath();
+    accessSync(bundled, fsConstants.X_OK);
+    return bundled;
+  } catch { /* no bundled browser */ }
+  for (const candidate of ["/usr/bin/google-chrome", "/usr/bin/google-chrome-stable", "/usr/bin/chromium", "/usr/bin/chromium-browser"]) {
+    try { accessSync(candidate, fsConstants.X_OK); return candidate; } catch { /* try next */ }
+  }
+  return null;
+}
+
+const executablePath = browserExecutable();
 const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
 const freePort = () => new Promise((resolve, reject) => {
   const server = createServer();
@@ -28,7 +45,10 @@ const waitFor = async (fn, label, timeout = 15_000) => {
   throw last instanceof Error ? last : new Error(`Timed out waiting for ${label}`);
 };
 
-test("browser silently reconnects a dropped terminal socket to the same live shell", { timeout: 45_000 }, async () => {
+test("browser silently reconnects a dropped terminal socket to the same live shell", {
+  timeout: 45_000,
+  skip: executablePath ? false : "No local Chrome executable; the mandatory transport/keepalive contract still runs.",
+}, async () => {
   const dataDir = mkdtempSync(join(tmpdir(), "1helm-terminal-reconnect-"));
   const appPort = await freePort();
   const mockPort = await freePort();
@@ -62,7 +82,7 @@ test("browser silently reconnects a dropped terminal socket to the same live she
     await api("/api/setup/complete", registration.token, { name: "Reconnect Test", terminals_enabled: true, provider_id: provider.provider.id, model: "mock-large" });
     const channel = (await api("/api/channels", registration.token, { name: "terminal-reconnect", purpose: "Prove terminal session continuity." })).channel;
 
-    browser = await puppeteer.launch({ headless: true, args: ["--no-sandbox", "--disable-setuid-sandbox"] });
+    browser = await puppeteer.launch({ executablePath, headless: true, args: ["--no-sandbox", "--disable-setuid-sandbox"] });
     const page = await browser.newPage();
     await page.evaluateOnNewDocument(() => {
       const NativeWebSocket = window.WebSocket;