diff --git a/docs/portals-and-agent-communication.md b/docs/portals-and-agent-communication.md index 8d40b23e2..a7b5e5448 100644 --- a/docs/portals-and-agent-communication.md +++ b/docs/portals-and-agent-communication.md @@ -37,6 +37,26 @@ WEBAPP_PORT=3300 INSTANT_PORT=5968 ``` +Host Portals survive gateway restarts, but are not permanent services: + +- Archiving a session stops its owned host Portals, including those in attached + repositories. A sibling session's Portal in a shared checkout is left alone. + The five-minute reaper also catches archived and deleted owners missed by an + interrupted cleanup. +- On Linux, a host Portal sleeps after 30 minutes without authenticated Portal + requests or an established connection to its service port. WebSockets count + as use; session-list and readiness polling do not. A gateway restart grants a + fresh idle window. If connection telemetry is unavailable, idle cleanup is + skipped rather than risking an active preview. Running owners are protected. + A page navigation wakes a sleeping Portal; background requests do not. + Archive and orphan cleanup stop Portals permanently, without automatic wake. +- New host processes are refused when available RAM falls below 5% or 2 GiB, + memory full-stall pressure reaches 10% over ten seconds, or the shared user + workload slice reaches 90% of its memory soft limit. The existing configurable + host Portal count cap and free-memory floor also apply. Existing, matching + Portals can still be reused. These are admission guards, not permission to + kill another session's active work. + Every listening `*_PORT` service is a Portal. Host services map to `https://:`; Sandbox services get an allocated route in 20000–27999 that relays over the Sandbox's authenticated outbound connection, diff --git a/packages/core/opensession-server/opensession.ts b/packages/core/opensession-server/opensession.ts index 66eddfcd0..1ac6d4c7a 100644 --- a/packages/core/opensession-server/opensession.ts +++ b/packages/core/opensession-server/opensession.ts @@ -99,6 +99,7 @@ import { findSession, findSessionAsync, getCachedSessions, + getSessionListSnapshotAsync, invalidateSessionsCache, reconcileRecoverableSafetyFences, recordRunOutcome, @@ -863,7 +864,7 @@ if (!g.__opensessionBooted) { // Portal processes survive a coordinator restart by design. Reconcile their // durable owner records immediately and keep reaping deleted-session husks. - startPortalReaper(getAllSessions); + startPortalReaper(getSessionListSnapshotAsync); startRunnerPortalReaper(); // Desk todo reminders: push + Slack DM when a remindAt passes (todos.ts) diff --git a/packages/core/opensession-server/src/server/archive.ts b/packages/core/opensession-server/src/server/archive.ts index ab6de7738..dd147a79f 100644 --- a/packages/core/opensession-server/src/server/archive.ts +++ b/packages/core/opensession-server/src/server/archive.ts @@ -66,6 +66,16 @@ function releasePreviewLease(sessionId: string): void { } } +async function stopArchivedPortals(sessionId: string): Promise { + try { + const { stopArchivedSessionPortals } = await import("./portal-supervisor"); + await stopArchivedSessionPortals(sessionId); + } catch (error) { + // Archiving remains committed. The periodic lifecycle reaper retries. + console.error(`[portals] archive cleanup failed for ${sessionId}:`, error); + } +} + function toEntry(raw: RawEntry): Entry { return typeof raw === "string" ? { at: raw, reason: "manual" } : raw; } @@ -126,6 +136,7 @@ export async function setArchived( if (archived) { releasePreviewLease(id); await unpinEverywhere([id]); + await stopArchivedPortals(id); } } @@ -155,6 +166,7 @@ export async function archiveOlderThan( releasePreviewLease(session.id); await setIndexedSessionArchived(session.id, true, "idle"); publishSessionRow(session.id); + await stopArchivedPortals(session.id); } // Registry is written, so isArchivedId now reflects this batch — drop the // stale session/alias pins and any workspace pin whose last session just went. diff --git a/packages/core/opensession-server/src/server/portal-lifecycle.test.ts b/packages/core/opensession-server/src/server/portal-lifecycle.test.ts new file mode 100644 index 000000000..d111f4077 --- /dev/null +++ b/packages/core/opensession-server/src/server/portal-lifecycle.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, test } from "bun:test"; +import { + establishedPorts, + HostPortalActivity, + PORTAL_IDLE_MS, + portalCapacityProblem, +} from "./portal-lifecycle"; + +const healthy = { + meminfo: "MemTotal: 100000000 kB\nMemAvailable: 50000000 kB\n", +}; + +describe("host Portal admission", () => { + test("allows healthy hosts with or without a configured cgroup budget", () => { + expect(portalCapacityProblem(healthy)).toBeNull(); + expect( + portalCapacityProblem({ ...healthy, current: "500", high: "max" }), + ).toBeNull(); + expect( + portalCapacityProblem({ ...healthy, current: "500", high: "1000" }), + ).toBeNull(); + }); + test("rejects memory starvation and malformed availability", () => { + expect( + portalCapacityProblem({ + meminfo: "MemTotal: 100000000 kB\nMemAvailable: 1000000 kB", + }), + ).toContain("nearly full"); + expect(portalCapacityProblem({ meminfo: "" })).toContain( + "could not be measured", + ); + }); + test("rejects sustained reclaim stalls, not old swap usage", () => { + expect( + portalCapacityProblem({ + ...healthy, + pressure: "some avg10=90.00\nfull avg10=10.00 avg60=4.00", + }), + ).toContain("stalled"); + expect( + portalCapacityProblem({ + ...healthy, + pressure: "some avg10=5.00\nfull avg10=0.00 avg60=90.00", + }), + ).toBeNull(); + }); + test("reserves headroom before the aggregate soft limit", () => { + expect( + portalCapacityProblem({ ...healthy, current: "899", high: "1000" }), + ).toBeNull(); + expect( + portalCapacityProblem({ ...healthy, current: "900", high: "1000" }), + ).toContain("preview memory budget"); + }); +}); + +describe("Portal idle activity", () => { + test("expiry counts from discovery, not an ancient process start time", () => { + const activity = new HostPortalActivity(); + activity.observe(4000, "old-process", 100); + expect(activity.idle(4000, "old-process", 100 + PORTAL_IDLE_MS - 1)).toBe( + false, + ); + expect(activity.idle(4000, "old-process", 100 + PORTAL_IDLE_MS)).toBe(true); + }); + test("HTTP traffic extends the window but repeated observation does not", () => { + const activity = new HostPortalActivity(); + activity.observe(4000, "a", 0); + activity.observe(4000, "a", PORTAL_IDLE_MS - 1); + expect(activity.idle(4000, "a", PORTAL_IDLE_MS)).toBe(true); + activity.touch(4000, PORTAL_IDLE_MS); + expect(activity.idle(4000, "a", 2 * PORTAL_IDLE_MS - 1)).toBe(false); + }); + test("a replacement process gets its own idle window", () => { + const activity = new HostPortalActivity(); + activity.observe(4000, "a", 0); + activity.observe(4000, "b", PORTAL_IDLE_MS); + expect(activity.idle(4000, "a", 2 * PORTAL_IDLE_MS)).toBe(false); + expect(activity.idle(4000, "b", PORTAL_IDLE_MS + 1)).toBe(false); + activity.retain(new Set()); + activity.touch(4000, 3 * PORTAL_IDLE_MS); + expect(activity.idle(4000, "b", 4 * PORTAL_IDLE_MS)).toBe(false); + }); + test("recognizes established IPv4/IPv6 server ports, not listeners or peers", () => { + expect( + establishedPorts([ + "sl local_address rem_address st\n0: 0100007F:0FA0 0100007F:CB22 01\n1: 0100007F:0FA1 00000000:0000 0A", + "sl local_address rem_address st\n0: 00000000000000000000000001000000:0FA2 00000000000000000000000001000000:CAAA 01\nmalformed", + ]), + ).toEqual(new Set([4000, 4002])); + }); +}); diff --git a/packages/core/opensession-server/src/server/portal-lifecycle.ts b/packages/core/opensession-server/src/server/portal-lifecycle.ts new file mode 100644 index 000000000..b962347fe --- /dev/null +++ b/packages/core/opensession-server/src/server/portal-lifecycle.ts @@ -0,0 +1,144 @@ +import { readFile } from "node:fs/promises"; + +/** An unused host Portal gets one quiet half-hour, including after gateway boot. */ +export const PORTAL_IDLE_MS = 30 * 60_000; + +/** Request timestamps are process-local. Rediscovery grants a fresh idle window + * rather than killing a browser's Portal using a pre-restart timestamp. */ +export class HostPortalActivity { + private readonly ports = new Map< + number, + { generation: string; lastUsedAt: number } + >(); + + observe(port: number, generation: string, now: number): void { + if (this.ports.get(port)?.generation !== generation) + this.ports.set(port, { generation, lastUsedAt: now }); + } + + touch(port: number, now = Date.now()): void { + const entry = this.ports.get(port); + if (entry) entry.lastUsedAt = now; + } + + lastUsedAt(port: number, generation: string): number | undefined { + const entry = this.ports.get(port); + return entry?.generation === generation ? entry.lastUsedAt : undefined; + } + + idle( + port: number, + generation: string, + now: number, + idleMs = PORTAL_IDLE_MS, + ): boolean { + const entry = this.ports.get(port); + return entry?.generation === generation && now - entry.lastUsedAt >= idleMs; + } + + retain(ports: ReadonlySet): void { + for (const port of this.ports.keys()) + if (!ports.has(port)) this.ports.delete(port); + } +} + +export const hostPortalActivity = new HostPortalActivity(); + +/** /proc reports server-side local ports too, so an established WebSocket + * protects an open preview even when no further HTTP auth probes arrive. */ +export function establishedPorts(tables: readonly string[]): Set { + const ports = new Set(); + for (const table of tables) { + for (const line of table.split("\n").slice(1)) { + const fields = line.trim().split(/\s+/); + if (fields[3] !== "01") continue; + const port = Number.parseInt(fields[1]?.split(":")[1] ?? "", 16); + if (Number.isInteger(port) && port > 0 && port <= 65535) ports.add(port); + } + } + return ports; +} + +async function optionalFile(path: string): Promise { + try { + return await readFile(path, "utf8"); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") + return undefined; + throw error; + } +} + +export async function activeHostPortalPorts(): Promise | null> { + if (process.platform !== "linux") return null; + try { + const tables = await Promise.all([ + optionalFile("/proc/net/tcp"), + optionalFile("/proc/net/tcp6"), + ]); + const available = tables.filter((table) => table !== undefined); + return available.length ? establishedPorts(available) : null; + } catch { + // No connection evidence means no idle termination. Archive/orphan cleanup + // does not depend on this probe and still runs. + return null; + } +} + +export function portalCapacityProblem(input: { + meminfo: string; + pressure?: string; + current?: string; + high?: string; +}): string | null { + const total = Number(input.meminfo.match(/^MemTotal:\s+(\d+)/m)?.[1]); + const available = Number(input.meminfo.match(/^MemAvailable:\s+(\d+)/m)?.[1]); + if (!Number.isFinite(total) || !Number.isFinite(available) || total <= 0) + return "host memory availability could not be measured"; + // Leave at least 2 GiB and 5% of RAM for the control plane and OS. + if (available < Math.max(2 * 1024 * 1024, total * 0.05)) + return "host memory is nearly full"; + const fullStall = Number( + input.pressure?.match(/^full\s+avg10=([\d.]+)/m)?.[1], + ); + if (fullStall >= 10) return "the host is stalled reclaiming memory"; + const current = Number(input.current?.trim()); + const high = Number(input.high?.trim()); + // Stop admitting more previews before the shared soft limit starts reclaim. + if ( + Number.isFinite(current) && + Number.isFinite(high) && + high > 0 && + current >= high * 0.9 + ) + return "the preview memory budget is nearly full"; + return null; +} + +let capacityProbe: (() => Promise) | null = null; + +/** Unit tests spawn real Portals; the live host's memory pressure must not + * decide whether they pass. */ +export function _setHostPortalCapacityProbeForTests( + probe: (() => Promise) | null, +): void { + capacityProbe = probe; +} + +export async function assertHostPortalCapacity(): Promise { + if (capacityProbe) return capacityProbe(); + if (process.platform !== "linux") return; + const uid = process.getuid?.() ?? 1000; + const slice = `/sys/fs/cgroup/user.slice/user-${uid}.slice/user@${uid}.service/opensession.slice`; + const [meminfo, pressure, current, high] = await Promise.all([ + readFile("/proc/meminfo", "utf8"), + optionalFile("/proc/pressure/memory"), + optionalFile(`${slice}/memory.current`), + optionalFile(`${slice}/memory.high`), + ]); + const problem = portalCapacityProblem({ meminfo, pressure, current, high }); + if (problem) + throw new Error( + `Cannot start Portal: ${problem}. Stop an unused Portal and try again.`, + ); +} diff --git a/packages/core/opensession-server/src/server/portal-supervisor.test.ts b/packages/core/opensession-server/src/server/portal-supervisor.test.ts index c9cca8365..9abbf254e 100644 --- a/packages/core/opensession-server/src/server/portal-supervisor.test.ts +++ b/packages/core/opensession-server/src/server/portal-supervisor.test.ts @@ -11,23 +11,34 @@ import { tmpdir } from "os"; import { join } from "path"; import { createServer } from "node:net"; import { + applyPortalRegistryWrites, listPortalServices, listSandboxPortalServices, hostPortalAdmissionReason, normalizePortalPath, portalShouldSleep, portalsNeedingContainment, + portalGeneration, type PortalRecord, portalsToRestore, readPortalRegistry, reapOrphanedPortalServices, + sleepIdlePortalServices, + wakeHostPortalRoute, + restartPortalService, SANDBOX_PORTAL_AGENT_ENTRY, setPortalPath, startPortalService, startSandboxPortalService, + stopArchivedSessionPortals, stopPortalService, stopSandboxPortalService, } from "./portal-supervisor"; +import { + _setHostPortalCapacityProbeForTests, + HostPortalActivity, + PORTAL_IDLE_MS, +} from "./portal-lifecycle"; import { sleepingSandboxPortalStatus } from "./sandbox-portals"; import type { Sandbox } from "./sandbox/provider"; @@ -65,10 +76,14 @@ if (!testSetsid) { beforeEach(() => { worktree = mkdtempSync(join(tmpdir(), "os-portals-test-")); process.env.OPENSESSION_STATE_DIR = worktree; + // Real Portals start below; the host running this suite may itself be + // under memory pressure, which must not decide the outcome. + _setHostPortalCapacityProbeForTests(async () => {}); // Lifecycle fixtures must run on small CI hosts; admission boundaries are tested separately. process.env.OPENSESSION_PORTAL_MIN_AVAILABLE_MEMORY_MB = "1"; }); afterAll(() => { + _setHostPortalCapacityProbeForTests(null); if (worktree) rmSync(worktree, { recursive: true, force: true }); if (previousStateDir == null) delete process.env.OPENSESSION_STATE_DIR; else process.env.OPENSESSION_STATE_DIR = previousStateDir; @@ -87,6 +102,350 @@ afterAll(() => { rmSync(processTools, { recursive: true, force: true }); }); +describe("host Portal lifecycle cleanup", () => { + function registry(owner: string | undefined = "owner") { + const records: PortalRecord[] = [ + { + name: "web", + key: "WEB_PORT", + command: "serve", + port: 18091, + state: "awake", + scopeUnit: "os-test-portal-never-running", + sessionId: owner, + startedAt: "2020-01-01T00:00:00Z", + }, + ]; + writeFileSync( + join(worktree, ".ports.conf"), + records + .map((record) => `# opensession-portal ${JSON.stringify(record)}`) + .join("\n"), + ); + } + + const owner = (archived = false) => ({ + id: "owner", + worktreeDir: worktree, + attachedRepos: [], + archived, + }); + + test("archived owners no longer protect their preview", async () => { + registry(); + const result = await reapOrphanedPortalServices([owner(true)]); + expect(result.stopped).toHaveLength(1); + expect(readPortalRegistry(worktree)[0]?.state).toBe("stopped"); + }); + + test("preserves a sibling owner's Portal in a shared worktree", async () => { + registry(); + const result = await reapOrphanedPortalServices([ + owner(), + { ...owner(true), id: "sibling" }, + ]); + expect(result.stopped).toEqual([]); + }); + + for (const attached of [false, true]) { + test(`alias-owned Portals follow canonical archive state in ${attached ? "attached" : "primary"} worktrees`, async () => { + const aliasId = "slack-C998-1719860000.000000"; + registry(aliasId); + const canonical = { + ...owner(), + aliasIds: [aliasId], + worktreeDir: attached ? "" : worktree, + attachedRepos: attached + ? [{ repo: "attached", branch: "main", dir: worktree }] + : [], + }; + expect((await reapOrphanedPortalServices([canonical])).stopped).toEqual( + [], + ); + expect(readPortalRegistry(worktree)[0]?.state).toBe("awake"); + expect( + (await reapOrphanedPortalServices([{ ...canonical, archived: true }])) + .stopped, + ).toHaveLength(1); + expect(readPortalRegistry(worktree)[0]?.state).toBe("stopped"); + }); + } + + test("alias-owned Portals retain idle expiry without sleeping a running owner", async () => { + registry("slack-alias"); + const canonical = { ...owner(), aliasIds: ["slack-alias"] }; + const activity = new HostPortalActivity(); + const sweep = (now: number, isRunning = false) => + sleepIdlePortalServices([{ ...canonical, isRunning }], { + now, + activity, + activePorts: new Set(), + }); + expect((await sweep(0)).slept).toEqual([]); + expect((await sweep(PORTAL_IDLE_MS, true)).slept).toEqual([]); + expect((await sweep(PORTAL_IDLE_MS)).slept).toHaveLength(1); + expect(readPortalRegistry(worktree)[0]?.state).toBe("sleeping"); + }); + + test("legacy ownerless Portals survive until every worktree owner archives", async () => { + registry(""); + expect( + ( + await reapOrphanedPortalServices([ + owner(), + { ...owner(true), id: "sibling" }, + ]) + ).stopped, + ).toEqual([]); + expect( + (await reapOrphanedPortalServices([owner(true)])).stopped, + ).toHaveLength(1); + }); + + test("sleeps unused Portals but preserves HTTP activity and established connections", async () => { + registry(); + const activity = new HostPortalActivity(); + const sweep = ( + now: number, + ports: ReadonlySet | null = new Set(), + ) => + sleepIdlePortalServices([owner()], { + now, + activity, + activePorts: ports, + }); + expect((await sweep(0)).slept).toEqual([]); + activity.touch(18091, PORTAL_IDLE_MS - 1); + expect((await sweep(PORTAL_IDLE_MS)).slept).toEqual([]); + expect((await sweep(2 * PORTAL_IDLE_MS, new Set([18091]))).slept).toEqual( + [], + ); + expect((await sweep(3 * PORTAL_IDLE_MS, null)).slept).toEqual([]); + expect((await sweep(3 * PORTAL_IDLE_MS)).slept).toHaveLength(1); + expect(readPortalRegistry(worktree)[0]?.state).toBe("sleeping"); + // The archive sweep must not turn an idle, wakeable Portal into stopped. + expect((await reapOrphanedPortalServices([owner()])).stopped).toEqual([]); + }); + + test("idle sleep stays wakeable and concurrent wakes reuse one replacement", async () => { + const started = await startPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "wakeable", + defaultPath: "/kept?mode=preview", + readyTimeoutMs: 15_000, + command: + "bun -e 'Bun.serve({port:Number(process.env.PORT),fetch(){return new Response(\"awake\")}})'", + }); + const port = started.port; + const activity = new HostPortalActivity(); + try { + expect( + ( + await sleepIdlePortalServices([owner()], { + now: 0, + activity, + activePorts: new Set(), + }) + ).slept, + ).toEqual([]); + expect( + ( + await sleepIdlePortalServices([owner()], { + now: PORTAL_IDLE_MS, + activity, + activePorts: new Set(), + }) + ).slept, + ).toHaveLength(1); + expect((await listPortalServices(worktree))[0]?.state).toBe("sleeping"); + const [first, second] = await Promise.all([ + wakeHostPortalRoute(port), + wakeHostPortalRoute(port), + ]); + expect(first.pid).toBe(second.pid); + expect(first.pid).not.toBe(started.pid); + expect(first.defaultPath).toBe("/kept?mode=preview"); + expect(first.readyTimeoutMs).toBe(15_000); + expect(await (await fetch(`http://127.0.0.1:${port}`)).text()).toBe( + "awake", + ); + } finally { + await stopPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "wakeable", + }); + } + }); + + test("a starved host refuses a fresh Portal and records why", async () => { + _setHostPortalCapacityProbeForTests(async () => { + throw new Error("Cannot start Portal: host memory is nearly full."); + }); + await expect( + startPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "web", + port: 18093, + command: "sleep 60", + }), + ).rejects.toThrow("host memory is nearly full"); + expect(readPortalRegistry(worktree)[0]).toMatchObject({ + name: "web", + state: "failed", + lastError: expect.stringContaining("host memory is nearly full"), + }); + }); + + test("archiving through an alias stops Portals under every id of the owner", async () => { + registry("canonical"); + const web = readPortalRegistry(worktree)[0]!; + writeFileSync( + join(worktree, ".ports.conf"), + [ + web, + { + ...web, + name: "api", + key: "API_PORT", + port: 18092, + sessionId: "slack-C998-1719860000.000000", + }, + { + ...web, + name: "other", + key: "OTHER_PORT", + port: 18093, + sessionId: "sibling", + }, + ] + .map((record) => `# opensession-portal ${JSON.stringify(record)}`) + .join("\n"), + ); + await stopArchivedSessionPortals("slack-C998-1719860000.000000", { + findSession: async () => ({ + id: "canonical", + aliasIds: ["slack-C998-1719860000.000000"], + worktreeDir: worktree, + attachedRepos: [], + }), + }); + expect( + readPortalRegistry(worktree).map((record) => [record.name, record.state]), + ).toEqual([ + ["web", "stopped"], + ["api", "stopped"], + ["other", "awake"], + ]); + }); + + test("a status probe of a replaced generation cannot mark the replacement failed", () => { + registry(); + const first = readPortalRegistry(worktree)[0]!; + const replacement = { + ...first, + pid: 4242, + startedAt: "2021-01-01T00:00:00Z", + }; + const probed = { + ...first, + state: "failed" as const, + lastError: "The service is no longer listening.", + }; + // The poll read the first incarnation; a restart installed the replacement + // before the poll's write landed. + expect(applyPortalRegistryWrites([replacement], [first], [probed])).toEqual( + [replacement], + ); + // The same write against the incarnation it probed still lands, and a + // locked start over a probe-failed record keeps its own generation. + expect(applyPortalRegistryWrites([first], [first], [probed])).toEqual([ + probed, + ]); + const starting = { ...first, state: "starting" as const, pid: undefined }; + const launched = { ...starting, pid: 4243 }; + expect( + applyPortalRegistryWrites( + [{ ...starting, state: "failed" }], + [starting], + [launched], + ), + ).toEqual([launched]); + }); + + test("a stale stop queued behind a restart cannot kill the replacement", async () => { + const port = 18094; + const first = await startPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "web", + port, + command: + "bun -e 'Bun.serve({port:Number(process.env.PORT),fetch(){return new Response(\"web\")}})'", + }); + const restart = restartPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "web", + }); + // The restart is inside its stop (SIGTERM plus a grace sleep) when the + // sweep's stop arrives holding the old generation. + await Bun.sleep(100); + const sweep = stopPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "web", + expectedGeneration: portalGeneration(first), + }); + const replacement = await restart; + await expect(sweep).rejects.toThrow("Portal changed"); + expect(replacement.state).toBe("awake"); + expect(replacement.pid).not.toBe(first.pid); + expect((await listPortalServices(worktree))[0]).toMatchObject({ + state: "awake", + pid: replacement.pid, + }); + await stopPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "web", + }); + }, 20_000); + + test("an expired generation cannot stop a replacement", async () => { + registry(); + await expect( + stopPortalService({ + sessionId: "owner", + worktreeDir: worktree, + name: "web", + expectedGeneration: "obsolete", + }), + ).rejects.toThrow("Portal changed"); + expect(readPortalRegistry(worktree)[0]?.state).toBe("awake"); + }); + + test("concurrent registry updates preserve unrelated services", async () => { + registry(); + const web = readPortalRegistry(worktree)[0]; + writeFileSync( + join(worktree, ".ports.conf"), + [web, { ...web, name: "api", key: "API_PORT", port: 18092 }] + .map((record) => `# opensession-portal ${JSON.stringify(record)}`) + .join("\n"), + ); + await Promise.all([ + setPortalPath(worktree, "/web", "web"), + setPortalPath(worktree, "/api", "api"), + ]); + expect( + readPortalRegistry(worktree).map((record) => record.defaultPath), + ).toEqual(["/web", "/api"]); + }); +}); + describe("portalsToRestore", () => { const record = ( name: string, @@ -231,7 +590,7 @@ describe("session Portal supervisor", () => { expect(existsSync(SANDBOX_PORTAL_AGENT_ENTRY)).toBe(true); }); - test("keeps generated portal metadata and ports together in .ports.conf", () => { + test("keeps generated portal metadata and ports together in .ports.conf", async () => { writeFileSync(join(worktree, ".ports.conf"), "WEBAPP_PORT=3300\n"); const record = { name: "api", @@ -244,7 +603,7 @@ describe("session Portal supervisor", () => { join(worktree, ".ports.conf"), `${PREFIX(record)}\nPORTAL_API_PORT=4200\nWEBAPP_PORT=3300\n`, ); - setPortalPath(worktree, "/health", "api"); + await setPortalPath(worktree, "/health", "api"); const [portal] = readPortalRegistry(worktree); expect(portal).toMatchObject({ name: "api", @@ -270,7 +629,7 @@ describe("session Portal supervisor", () => { `\x1b]0;@modal: cat .ports.conf\x07${PREFIX(record)}\nWEBAPP_PORT=4000\n`, ); - setPortalPath(worktree, "/videos", "web"); + await setPortalPath(worktree, "/videos", "web"); const text = await Bun.file(join(worktree, ".ports.conf")).text(); expect(text).not.toContain("\x1b"); diff --git a/packages/core/opensession-server/src/server/portal-supervisor.ts b/packages/core/opensession-server/src/server/portal-supervisor.ts index 25907198f..e8aca3e82 100644 --- a/packages/core/opensession-server/src/server/portal-supervisor.ts +++ b/packages/core/opensession-server/src/server/portal-supervisor.ts @@ -6,17 +6,24 @@ * agent can inspect services without becoming their process manager. */ +import { existsSync, readFileSync } from "fs"; import { - closeSync, - existsSync, - mkdirSync, - openSync, - readdirSync, - readFileSync, - realpathSync, - writeFileSync, -} from "fs"; + mkdir, + open, + readFile, + readdir, + rename, + rm, + realpath, + writeFile, +} from "node:fs/promises"; import { join, resolve } from "path"; +import { + activeHostPortalPorts, + assertHostPortalCapacity, + hostPortalActivity, + type HostPortalActivity, +} from "./portal-lifecycle"; import { audit } from "./audit"; import { ensureAgentAwsCredsFile } from "./aws-creds"; import { configuredPaths, configuredServer } from "./config"; @@ -115,7 +122,6 @@ type HostPortalRef = { const portalGlobal = globalThis as unknown as { __opensessionHostPortalRefs?: Map; - __opensessionHostPortalAccess?: Map; __opensessionHostPortalWakes?: Map< string, Promise @@ -123,8 +129,6 @@ const portalGlobal = globalThis as unknown as { __opensessionHostPortalReservations?: Set; }; const hostPortalRefs = (portalGlobal.__opensessionHostPortalRefs ??= new Map()); -const hostPortalAccess = (portalGlobal.__opensessionHostPortalAccess ??= - new Map()); const hostPortalWakes = (portalGlobal.__opensessionHostPortalWakes ??= new Map()); const hostPortalReservations = @@ -205,6 +209,34 @@ export function readPortalRegistry(worktreeDir: string): PortalRecord[] { : []; } +async function readHostRegistryText(worktreeDir: string): Promise { + try { + return await readFile(registryPath(worktreeDir), "utf8"); + } catch (error) { + if (error instanceof Error && "code" in error && error.code === "ENOENT") + return ""; + throw error; + } +} + +async function readHostPortalRegistry( + worktreeDir: string, +): Promise { + return parsePortalRegistry(await readHostRegistryText(worktreeDir)); +} + +/** Identifies one process incarnation of a Portal record, so a cleanup + * decided against an old incarnation cannot act on its replacement. */ +export function portalGeneration(portal: PortalRecord): string { + return JSON.stringify([ + portal.sessionId, + portal.name, + portal.startedAt, + portal.pid, + portal.scopeUnit, + ]); +} + function serializedPortalRegistry( previousText: string, records: PortalRecord[], @@ -224,13 +256,91 @@ function serializedPortalRegistry( return [...kept, ...generated, ""].join("\n"); } -function writePortalRegistry( +const hostRegistryWrites = new Map>(); +const hostPortalOperations = new Map>(); + +/** + * Serialize start, stop, and restart of one host Portal by canonical worktree + * and name. A stop validates the record's generation, terminates the process + * group (the scope unit is stable per worktree and name), and persists the + * stopped record; none of that may interleave with a start or restart that + * replaces the record, or the stale stop kills the replacement and overwrites + * its record. Unlike the Sandbox seam this queues rather than coalesces: a + * stop after a start must run after it, not return the start's result. + */ +async function withHostPortalOperation( worktreeDir: string, - records: PortalRecord[], -): void { - const path = registryPath(worktreeDir); - const previous = existsSync(path) ? readFileSync(path, "utf8") : ""; - writeFileSync(path, serializedPortalRegistry(previous, records)); + name: string, + operation: () => Promise, +): Promise { + const key = `${await canonicalDir(worktreeDir)}:${name}`; + const previous = hostPortalOperations.get(key) ?? Promise.resolve(); + const task = previous.catch(() => {}).then(operation); + hostPortalOperations.set(key, task); + try { + return await task; + } finally { + if (hostPortalOperations.get(key) === task) + hostPortalOperations.delete(key); + } +} + +async function updateHostPortalRegistry( + worktreeDir: string, + update: (records: PortalRecord[]) => PortalRecord[], +): Promise { + const dir = await canonicalDir(worktreeDir); + const previous = hostRegistryWrites.get(dir) ?? Promise.resolve(); + const next = previous + .catch(() => {}) + .then(async () => { + const text = await readHostRegistryText(dir); + const records = update(parsePortalRegistry(text)); + const path = registryPath(dir); + const temporary = `${path}.${process.pid}.${crypto.randomUUID()}.tmp`; + try { + await writeFile(temporary, serializedPortalRegistry(text, records)); + await rename(temporary, path); + } finally { + await rm(temporary, { force: true }); + } + }); + hostRegistryWrites.set(dir, next); + try { + await next; + } finally { + if (hostRegistryWrites.get(dir) === next) hostRegistryWrites.delete(dir); + } +} + +/** + * Merge one operation's registry writes into the registry as it is now. + * Only records this operation changed are applied: another Portal in a shared + * checkout may have started while this one waited for readiness. A change + * decided against one incarnation is dropped when the record has since been + * replaced. Start, stop, sleep, and restart change generation fields only + * under the Portal's operation lock, so their writes always land; the unlocked + * status poll only changes state, so a probe of generation A that finishes + * after a restart installed B cannot mark B failed and strand its process. + */ +export function applyPortalRegistryWrites( + latest: PortalRecord[], + snapshot: readonly PortalRecord[], + records: readonly PortalRecord[], +): PortalRecord[] { + for (const record of records) { + const before = snapshot.find((entry) => entry.name === record.name); + if (JSON.stringify(record) === JSON.stringify(before)) continue; + const current = latest.find((entry) => entry.name === record.name); + if ( + before && + current && + portalGeneration(current) !== portalGeneration(before) + ) + continue; + latest = upsert(latest, record); + } + return latest; } async function portListening(port: number): Promise { @@ -272,9 +382,19 @@ type PortalOps = { }; function hostPortalOps(worktreeDir: string): PortalOps { + let readSnapshot: PortalRecord[] = []; return { - readRegistry: async () => readPortalRegistry(worktreeDir), - writeRegistry: async (records) => writePortalRegistry(worktreeDir, records), + readRegistry: async () => { + readSnapshot = await readHostPortalRegistry(worktreeDir); + return readSnapshot; + }, + writeRegistry: async (records) => { + const snapshot = readSnapshot; + await updateHostPortalRegistry(worktreeDir, (latest) => + applyPortalRegistryWrites(latest, snapshot, records), + ); + readSnapshot = records; + }, probePort: portListening, pidAlive, scopeAlive: userScopeActive, @@ -329,7 +449,7 @@ async function waitForPortalPort( async function allocatePort(worktreeDir: string): Promise { const reserved = new Set( - readPortalRegistry(worktreeDir).map((record) => record.port), + (await readHostPortalRegistry(worktreeDir)).map((record) => record.port), ); for (let port = 4_000; port < 9_000; port++) { if (!reserved.has(port) && !(await portListening(port))) return port; @@ -590,10 +710,19 @@ async function terminatePortalProcess( if (await ops.pidAlive(pid)) await ops.signalGroup(pid, "SIGKILL"); } -async function stopPortal(ops: PortalOps, name: string): Promise { +async function stopPortal( + ops: PortalOps, + name: string, + expectedGeneration?: string, +): Promise { const records = await ops.readRegistry(); const current = records.find((record) => record.name === name); if (!current) throw new Error(`Portal '${name}' does not exist.`); + if ( + expectedGeneration !== undefined && + portalGeneration(current) !== expectedGeneration + ) + throw new Error("Portal changed before cleanup; retry on the next sweep."); await terminatePortalProcess(ops, current); const stopped = { ...current, @@ -625,11 +754,11 @@ export async function listPortalServices( worktreeDir: string, ): Promise { const records = await listPortals(hostPortalOps(worktreeDir)); - for (const record of records) registerHostPortal(worktreeDir, record); + for (const record of records) await registerHostPortal(worktreeDir, record); return records; } -export async function startPortalService(input: { +type HostPortalStartInput = { sessionId: string; worktreeDir: string; name: string; @@ -641,14 +770,22 @@ export async function startPortalService(input: { readyTimeoutMs?: number; /** Narrow, caller-owned additions for a trusted declared recipe. */ env?: Record; -}): Promise { - const current = readPortalRegistry(input.worktreeDir).find( - (record) => record.name === input.name, +}; + +export function startPortalService( + input: HostPortalStartInput, +): Promise { + return withHostPortalOperation( + input.worktreeDir, + validateName(input.name), + () => startHostPortal(input), ); - const releaseAdmission = - current?.state === "awake" - ? () => {} - : reserveHostPortalStart(input.worktreeDir, input.name); +} + +/** The start itself; callers hold this Portal's operation lock. */ +async function startHostPortal( + input: HostPortalStartInput, +): Promise { const logDir = join(sessionScratchRoot(), input.sessionId, "portals"); const logPath = join(logDir, `${input.name}.log`); // The same short-lived AWS credentials the agent's own shell gets (a pointer @@ -656,6 +793,7 @@ export async function startPortalService(input: { // repository's dev server (tella-fusion's start.sh) otherwise falls back to // an operator SSO profile and hangs on an interactive login. {} when the // mint is off. + let releaseAdmission: () => Promise = async () => {}; try { const awsEnv = await ensureAgentAwsCredsFile(); const started = await startPortal(hostPortalOps(input.worktreeDir), { @@ -666,8 +804,13 @@ export async function startPortalService(input: { urlFor: (port) => `https://${configuredServer().previewHost}:${port + 6_000}`, launch: async ({ name, command, port, url }) => { - mkdirSync(logDir, { recursive: true }); - const log = openSync(logPath, "w"); + await assertHostPortalCapacity(); + releaseAdmission = await reserveHostPortalStart( + input.worktreeDir, + name, + ); + await mkdir(logDir, { recursive: true }); + const log = await open(logPath, "w"); const directCommand = ["setsid", "bash", "-lc", `exec ${command}`]; const portalEnv = { PATH: process.env.PATH || "/usr/local/bin:/usr/bin:/bin", @@ -687,16 +830,20 @@ export async function startPortalService(input: { name, { env: portalEnv }, ); - const proc = Bun.spawn(scoped.command, { - cwd: input.worktreeDir, - // Portal commands are user-authored code. Do not hand them the Open - // Session service environment, which can include operator credentials. - env: scoped.env, - stdin: "ignore", - stdout: log, - stderr: log, - }); - closeSync(log); + let proc: ReturnType; + try { + proc = Bun.spawn(scoped.command, { + cwd: input.worktreeDir, + // Portal commands are user-authored code. Do not hand them the Open + // Session service environment, which can include operator credentials. + env: scoped.env, + stdin: "ignore", + stdout: log.fd, + stderr: log.fd, + }); + } finally { + await log.close(); + } proc.unref(); return { pid: proc.pid, @@ -704,7 +851,13 @@ export async function startPortalService(input: { }; }, }); - registerHostPortal(input.worktreeDir, started, true); + hostPortalActivity.observe( + started.port, + portalGeneration(started), + Date.now(), + ); + hostPortalActivity.touch(started.port); + await registerHostPortal(input.worktreeDir, started, true); audit({ msg: "portal_started", session_id: input.sessionId, @@ -713,18 +866,36 @@ export async function startPortalService(input: { }); return started; } finally { - releaseAdmission(); + await releaseAdmission(); } } -export async function stopPortalService(input: { +type HostPortalStopInput = { sessionId: string; worktreeDir: string; name: string; -}): Promise { + /** Refuse to stop a record that no longer matches `portalGeneration`. */ + expectedGeneration?: string; +}; + +export function stopPortalService( + input: HostPortalStopInput, +): Promise { + return withHostPortalOperation( + input.worktreeDir, + validateName(input.name), + () => stopHostPortal(input), + ); +} + +/** The stop itself; callers hold this Portal's operation lock. */ +async function stopHostPortal( + input: HostPortalStopInput, +): Promise { const stopped = await stopPortal( hostPortalOps(input.worktreeDir), validateName(input.name), + input.expectedGeneration, ); audit({ msg: "portal_stopped", @@ -739,32 +910,37 @@ async function sleepPortalService(input: { sessionId: string; worktreeDir: string; name: string; + generation: string; lastAccessedAt: number; + stillIdle: () => boolean; }): Promise { - const ops = hostPortalOps(input.worktreeDir); - const records = await ops.readRegistry(); - const current = records.find((record) => record.name === input.name); - if (!current) throw new Error(`Portal '${input.name}' does not exist.`); - if (current.state !== "awake") return current; - await terminatePortalProcess(ops, current); - const now = new Date().toISOString(); - const sleeping = { - ...current, - state: "sleeping" as const, - pid: undefined, - scopeUnit: undefined, - lastAccessedAt: new Date(input.lastAccessedAt).toISOString(), - sleptAt: now, - }; - await ops.writeRegistry(upsert(records, sleeping)); - registerHostPortal(input.worktreeDir, sleeping); - audit({ - msg: "portal_slept", - session_id: input.sessionId, - portal: sleeping.name, - port: sleeping.port, + return withHostPortalOperation(input.worktreeDir, input.name, async () => { + const ops = hostPortalOps(input.worktreeDir); + const records = await ops.readRegistry(); + const current = records.find((record) => record.name === input.name); + if (!current) throw new Error(`Portal '${input.name}' does not exist.`); + if (current.state !== "awake") return current; + if (portalGeneration(current) !== input.generation || !input.stillIdle()) + return current; + await terminatePortalProcess(ops, current); + const sleeping = { + ...current, + state: "sleeping" as const, + pid: undefined, + scopeUnit: undefined, + lastAccessedAt: new Date(input.lastAccessedAt).toISOString(), + sleptAt: new Date().toISOString(), + }; + await ops.writeRegistry(upsert(records, sleeping)); + await registerHostPortal(input.worktreeDir, sleeping); + audit({ + msg: "portal_slept", + session_id: input.sessionId, + portal: sleeping.name, + port: sleeping.port, + }); + return sleeping; }); - return sleeping; } /** Stop every host-managed Portal before its session workspace is removed. */ @@ -772,7 +948,7 @@ export async function stopAllPortalServices(input: { sessionId: string; worktreeDir: string; }): Promise { - const records = readPortalRegistry(input.worktreeDir); + const records = await readHostPortalRegistry(input.worktreeDir); for (const record of records) { if (record.state === "stopped") continue; try { @@ -788,9 +964,15 @@ export async function stopAllPortalServices(input: { export type PortalOwnerSession = Pick< UnifiedSession, - "id" | "worktreeDir" | "attachedRepos" + "id" | "worktreeDir" | "attachedRepos" | "aliasIds" > & - Partial>; + Partial>; +function portalOwnerIds( + session: Pick, +): string[] { + return [session.id, ...(session.aliasIds ?? [])]; +} + export type PortalReapResult = { stopped: Array<{ sessionId: string; worktreeDir: string; name: string }>; }; @@ -801,13 +983,30 @@ export type PortalSleepResult = { export async function sleepIdlePortalServices( sessions: readonly PortalOwnerSession[], - options: { now?: number; idleMs?: number } = {}, + options: { + now?: number; + idleMs?: number; + activity?: HostPortalActivity; + activePorts?: ReadonlySet | null; + } = {}, ): Promise { const now = options.now ?? Date.now(); const idleMs = options.idleMs ?? positiveIntegerEnv("OPENSESSION_PORTAL_IDLE_MS", DEFAULT_PORTAL_IDLE_MS); - const owners = new Map(sessions.map((session) => [session.id, session])); + const activity = options.activity ?? hostPortalActivity; + const activePorts = + options.activePorts === undefined + ? await activeHostPortalPorts() + : options.activePorts; + if (activePorts === null) return { slept: [] }; + // The catalog deliberately omits live runner overlays. Probe only Portal + // owners in the in-memory engine registry, never every historical actor. + const { isAgentEngineBusy } = await import("./agent-runner"); + const owners = new Map(); + for (const session of sessions) { + for (const id of portalOwnerIds(session)) owners.set(id, session); + } const ownerDirs = sessions .flatMap((session) => [ session.worktreeDir, @@ -815,20 +1014,19 @@ export async function sleepIdlePortalServices( ]) .filter((dir): dir is string => typeof dir === "string"); const slept: PortalSleepResult["slept"] = []; - for (const { ref, record } of managedHostPortals(ownerDirs)) { + const observedPorts = new Set(); + for (const { ref, record } of await managedHostPortals(ownerDirs)) { + observedPorts.add(record.port); const owner = owners.get(ref.sessionId); - if (!owner) continue; // The orphan reaper owns this case. - const key = portalRefKey(ref.worktreeDir, ref.name); - const persistedAccess = Date.parse( - record.lastAccessedAt ?? record.startedAt ?? "", - ); - const lastAccessedAt = - hostPortalAccess.get(key) ?? - (Number.isFinite(persistedAccess) ? persistedAccess : now); + if (!owner || owner.archived) continue; // The orphan/archive reaper owns these. + const generation = portalGeneration(record); + activity.observe(record.port, generation, now); + if (activePorts.has(record.port)) activity.touch(record.port, now); + const lastAccessedAt = activity.lastUsedAt(record.port, generation) ?? now; if ( !portalShouldSleep({ state: record.state, - ownerRunning: !!owner.isRunning, + ownerRunning: !!owner.isRunning || isAgentEngineBusy(owner.id), now, lastAccessedAt, idleMs, @@ -836,17 +1034,22 @@ export async function sleepIdlePortalServices( ) continue; try { - await sleepPortalService({ + const result = await sleepPortalService({ sessionId: ref.sessionId, worktreeDir: ref.worktreeDir, name: ref.name, + generation, lastAccessedAt, + stillIdle: () => + !isAgentEngineBusy(owner.id) && + activity.idle(record.port, generation, now, idleMs), }); - slept.push({ - sessionId: ref.sessionId, - worktreeDir: ref.worktreeDir, - name: ref.name, - }); + if (result.state === "sleeping") + slept.push({ + sessionId: ref.sessionId, + worktreeDir: ref.worktreeDir, + name: ref.name, + }); } catch (error) { console.warn( `[portals] could not sleep idle ${ref.name} in ${ref.worktreeDir}:`, @@ -854,6 +1057,7 @@ export async function sleepIdlePortalServices( ); } } + activity.retain(observedPorts); return { slept }; } @@ -880,16 +1084,17 @@ export type PortalContainmentMigrationResult = { * spelling, the same registry read under the alias made every Portal owned by * a session spelled the other way look orphaned, and the reaper killed it. */ -function canonicalDir(dir: string): string { +async function canonicalDir(dir: string): Promise { try { - return realpathSync(dir); + return await realpath(dir); } catch { return resolve(dir); } } function portalRefKey(worktreeDir: string, name: string): string { - return `${canonicalDir(worktreeDir)}\0${name}`; + // Callers register canonical directories before constructing a key. + return `${worktreeDir}\0${name}`; } function positiveIntegerEnv(name: string, fallback: number): number { @@ -897,58 +1102,60 @@ function positiveIntegerEnv(name: string, fallback: number): number { return Number.isSafeInteger(value) && value > 0 ? value : fallback; } -function registerHostPortal( +async function registerHostPortal( worktreeDir: string, portal: PortalRecord, accessedNow = false, -): void { +): Promise { if (!portal.sessionId) return; const ref = { sessionId: portal.sessionId, - worktreeDir: canonicalDir(worktreeDir), + worktreeDir: await canonicalDir(worktreeDir), name: portal.name, port: portal.port, }; hostPortalRefs.set(portal.port, ref); - const key = portalRefKey(ref.worktreeDir, ref.name); if (accessedNow) { - hostPortalAccess.set(key, Date.now()); - } else if (!hostPortalAccess.has(key)) { - const persisted = Date.parse( - portal.lastAccessedAt ?? portal.startedAt ?? "", - ); - hostPortalAccess.set( - key, - Number.isFinite(persisted) ? persisted : Date.now(), + hostPortalActivity.observe( + portal.port, + portalGeneration(portal), + Date.now(), ); + hostPortalActivity.touch(portal.port); } } -function managedHostPortals(additionalDirs: readonly string[] = []): Array<{ - ref: HostPortalRef; - record: PortalRecord; -}> { - const dirs = new Set(additionalDirs.filter(Boolean).map(canonicalDir)); +async function managedHostPortals( + additionalDirs: readonly string[] = [], +): Promise< + Array<{ + ref: HostPortalRef; + record: PortalRecord; + }> +> { + const dirs = new Set( + await Promise.all(additionalDirs.filter(Boolean).map(canonicalDir)), + ); for (const ref of hostPortalRefs.values()) dirs.add(ref.worktreeDir); try { - for (const entry of readdirSync(configuredPaths().worktreesDir, { + for (const entry of await readdir(configuredPaths().worktreesDir, { withFileTypes: true, })) { if (entry.isDirectory()) dirs.add( - canonicalDir(join(configuredPaths().worktreesDir, entry.name)), + await canonicalDir(join(configuredPaths().worktreesDir, entry.name)), ); } } catch {} const portals: Array<{ ref: HostPortalRef; record: PortalRecord }> = []; for (const worktreeDir of dirs) { - for (const record of readPortalRegistry(worktreeDir)) { + for (const record of await readHostPortalRegistry(worktreeDir)) { if ( !record.sessionId || (!record.scopeUnit && record.state !== "sleeping") ) continue; - registerHostPortal(worktreeDir, record); + await registerHostPortal(worktreeDir, record); portals.push({ ref: { sessionId: record.sessionId, @@ -963,9 +1170,9 @@ function managedHostPortals(additionalDirs: readonly string[] = []): Array<{ return portals; } -function hostPortalRecord( +async function hostPortalRecord( sourcePort: number, -): { ref: HostPortalRef; record: PortalRecord } | undefined { +): Promise<{ ref: HostPortalRef; record: PortalRecord } | undefined> { if ( !Number.isInteger(sourcePort) || sourcePort < MIN_PORT || @@ -974,11 +1181,11 @@ function hostPortalRecord( return; let ref = hostPortalRefs.get(sourcePort); if (!ref) { - managedHostPortals(); + await managedHostPortals(); ref = hostPortalRefs.get(sourcePort); } if (!ref) return; - const record = readPortalRegistry(ref.worktreeDir).find( + const record = (await readHostPortalRegistry(ref.worktreeDir)).find( (candidate) => candidate.name === ref!.name && candidate.port === sourcePort, ); @@ -990,28 +1197,28 @@ function hostPortalRecord( } /** Resolve one Caddy host-Portal upstream and record real HTTP activity. */ -export function hostPortalRouteStatus( +export async function hostPortalRouteStatus( sourcePort: number, -): { state: PortalState; sessionId: string } | undefined { - const found = hostPortalRecord(sourcePort); +): Promise<{ state: PortalState; sessionId: string } | undefined> { + const found = await hostPortalRecord(sourcePort); if (!found) return; - if (found.record.state === "awake") - hostPortalAccess.set( - portalRefKey(found.ref.worktreeDir, found.ref.name), - Date.now(), - ); + hostPortalActivity.observe( + found.record.port, + portalGeneration(found.record), + Date.now(), + ); return { state: found.record.state, sessionId: found.ref.sessionId }; } /** Wake a Portal that the idle reaper deliberately slept. Concurrent browser * refreshes share one cold start rather than spawning competing dev servers. */ -export function wakeHostPortalRoute( +export async function wakeHostPortalRoute( sourcePort: number, ): Promise { - const found = hostPortalRecord(sourcePort); + const found = await hostPortalRecord(sourcePort); if (!found) return Promise.reject(new Error("Host Portal is not registered")); if (found.record.state === "awake") { - registerHostPortal(found.ref.worktreeDir, found.record, true); + await registerHostPortal(found.ref.worktreeDir, found.record, true); return Promise.resolve({ ...found.record, url: `https://${configuredServer().previewHost}:${sourcePort + 6_000}`, @@ -1024,17 +1231,32 @@ export function wakeHostPortalRoute( const key = portalRefKey(found.ref.worktreeDir, found.ref.name); const existing = hostPortalWakes.get(key); if (existing) return existing; - const wake = startPortalService({ - sessionId: found.ref.sessionId, - worktreeDir: found.ref.worktreeDir, - name: found.record.name, - command: found.record.command, - port: found.record.port, - key: found.record.key, - description: found.record.description, - defaultPath: found.record.defaultPath, - readyTimeoutMs: found.record.readyTimeoutMs ?? 180_000, - }).finally(() => hostPortalWakes.delete(key)); + const wake = withHostPortalOperation( + found.ref.worktreeDir, + found.ref.name, + async () => { + const current = ( + await readHostPortalRegistry(found.ref.worktreeDir) + ).find((record) => record.name === found.ref.name); + if ( + !current || + current.state !== "sleeping" || + portalGeneration(current) !== portalGeneration(found.record) + ) + throw new Error("Portal changed before wake; reload its status."); + return startHostPortal({ + sessionId: found.ref.sessionId, + worktreeDir: found.ref.worktreeDir, + name: current.name, + command: current.command, + port: current.port, + key: current.key, + description: current.description, + defaultPath: current.defaultPath, + readyTimeoutMs: current.readyTimeoutMs ?? 180_000, + }); + }, + ).finally(() => hostPortalWakes.delete(key)); hostPortalWakes.set(key, wake); return wake; } @@ -1053,10 +1275,10 @@ export function hostPortalAdmissionReason(input: { return undefined; } -function availableMemoryMb(): number { +async function availableMemoryMb(): Promise { try { const match = /^MemAvailable:\s+(\d+)\s+kB$/m.exec( - readFileSync("/proc/meminfo", "utf8"), + await readFile("/proc/meminfo", "utf8"), ); return match ? Number(match[1]) / 1024 : Number.POSITIVE_INFINITY; } catch { @@ -1064,28 +1286,47 @@ function availableMemoryMb(): number { } } -function reserveHostPortalStart(worktreeDir: string, name: string): () => void { - const key = portalRefKey(worktreeDir, name); - const active = managedHostPortals().filter(({ record }) => - ["starting", "waking", "awake"].includes(record.state), - ).length; - const reason = hostPortalAdmissionReason({ - active, - reserved: hostPortalReservations.size, - maxActive: positiveIntegerEnv( - "OPENSESSION_MAX_HOST_PORTALS", - DEFAULT_MAX_HOST_PORTALS, - ), - availableMemoryMb: availableMemoryMb(), - minAvailableMemoryMb: positiveIntegerEnv( - "OPENSESSION_PORTAL_MIN_AVAILABLE_MEMORY_MB", - DEFAULT_MIN_AVAILABLE_MEMORY_MB, - ), +async function reserveHostPortalStart( + worktreeDir: string, + name: string, +): Promise<() => Promise> { + const root = configuredPaths().worktreesDir; + // Acquisitions and releases share a lane so a readiness completion cannot + // remove a reservation while an asynchronous census still sees the old tree. + return withHostPortalOperation(root, "@admission", async () => { + const key = portalRefKey(await canonicalDir(worktreeDir), name); + const portals = await managedHostPortals(); + const memoryMb = await availableMemoryMb(); + const active = new Set( + portals + .filter(({ record }) => + ["starting", "waking", "awake"].includes(record.state), + ) + .map(({ ref }) => portalRefKey(ref.worktreeDir, ref.name)), + ); + for (const reservation of hostPortalReservations) active.add(reservation); + const reason = hostPortalAdmissionReason({ + active: active.size, + maxActive: positiveIntegerEnv( + "OPENSESSION_MAX_HOST_PORTALS", + DEFAULT_MAX_HOST_PORTALS, + ), + availableMemoryMb: memoryMb, + minAvailableMemoryMb: positiveIntegerEnv( + "OPENSESSION_PORTAL_MIN_AVAILABLE_MEMORY_MB", + DEFAULT_MIN_AVAILABLE_MEMORY_MB, + ), + }); + if (reason) + throw new Error( + `${reason}. Stop another Portal or wait for it to sleep.`, + ); + hostPortalReservations.add(key); + return () => + withHostPortalOperation(root, "@admission", async () => { + hostPortalReservations.delete(key); + }); }); - if (reason) - throw new Error(`${reason}. Stop another Portal or wait for it to sleep.`); - hostPortalReservations.add(key); - return () => hostPortalReservations.delete(key); } export function portalShouldSleep(input: { @@ -1103,61 +1344,73 @@ export function portalShouldSleep(input: { } /** - * Stop host Portal process groups that no live session owns. Portal records - * are intentionally stored with the worktree, which survives a coordinator - * restart, so this closes the gap between a crashed delete and the next human - * action. Legacy records without sessionId are only reaped when no session - * owns their worktree at all. + * Reconcile host Portal lifetime: missing or archived owners and idle services. + * Durable records survive coordinator restarts. Legacy ownerless records are + * considered archived only when every owner of their worktree is archived. */ export async function reapOrphanedPortalServices( sessions: readonly PortalOwnerSession[], ): Promise { - const owners = new Map>(); - const addOwner = (dir: string | null | undefined, sessionId: string) => { + const owners = new Map>(); + const addOwner = async ( + dir: string | null | undefined, + session: PortalOwnerSession, + ) => { if (!dir) return; - const key = canonicalDir(dir); - const set = owners.get(key) ?? new Set(); - set.add(sessionId); + const key = await canonicalDir(dir); + const set = owners.get(key) ?? new Map(); + for (const id of portalOwnerIds(session)) + set.set(id, session.archived === true); owners.set(key, set); }; for (const session of sessions) { - addOwner(session.worktreeDir, session.id); + await addOwner(session.worktreeDir, session); for (const repo of session.attachedRepos ?? []) - addOwner(repo.dir, session.id); + await addOwner(repo.dir, session); } - // Include session worktrees outside the normal worktree root, then discover - // deleted-session worktrees below the managed root. We only act on explicit - // OpenSession Portal records, never arbitrary processes in those directories. + // Discover deleted-session worktrees too, but act only on explicit Portal + // records. Never enumerate or open session actor databases here. const dirs = new Set(owners.keys()); try { - for (const entry of readdirSync(configuredPaths().worktreesDir, { + for (const entry of await readdir(configuredPaths().worktreesDir, { withFileTypes: true, })) { if (entry.isDirectory()) dirs.add( - canonicalDir(join(configuredPaths().worktreesDir, entry.name)), + await canonicalDir(join(configuredPaths().worktreesDir, entry.name)), ); } } catch {} const stopped: PortalReapResult["stopped"] = []; for (const worktreeDir of dirs) { - const liveOwners = owners.get(worktreeDir) ?? new Set(); - for (const portal of readPortalRegistry(worktreeDir)) { + const liveOwners = owners.get(worktreeDir) ?? new Map(); + for (const portal of await readHostPortalRegistry(worktreeDir)) { if (portal.state === "stopped" || portal.state === "failed") continue; + const generation = portalGeneration(portal); const orphaned = portal.sessionId ? !liveOwners.has(portal.sessionId) : liveOwners.size === 0; - if (!orphaned) continue; + const archived = portal.sessionId + ? liveOwners.get(portal.sessionId) === true + : liveOwners.size > 0 && [...liveOwners.values()].every(Boolean); + const reason = orphaned ? "orphaned" : archived ? "archived" : null; + if (!reason) continue; const sessionId = portal.sessionId || "orphaned-portal"; try { - await stopPortalService({ sessionId, worktreeDir, name: portal.name }); + await stopPortalService({ + sessionId, + worktreeDir, + name: portal.name, + expectedGeneration: generation, + }); stopped.push({ sessionId, worktreeDir, name: portal.name }); audit({ - msg: "portal_orphan_reaped", + msg: "portal_reaped", session_id: sessionId, portal: portal.name, + reason, }); } catch (error) { console.warn( @@ -1170,6 +1423,67 @@ export async function reapOrphanedPortalServices( return { stopped }; } +/** Archive only the named session's services, never a sibling's Portal in a + * shared checkout. Legacy ownerless records are handled by the full reaper. */ +export async function stopArchivedSessionPortals( + sessionId: string, + options: { + findSession?: (id: string) => Promise; + } = {}, +): Promise { + const findSession = options.findSession ?? findMergedSession; + const session = await findSession(sessionId); + if (!session || session.runner || session.sandbox?.sandboxId) return; + // A Portal record carries the id its session ran under, which may be the + // canonical id or an alias merged into it. Every spelling owns the Portal. + const ownerIds = new Set(portalOwnerIds(session)); + const dirs = new Set([ + session.worktreeDir, + ...(session.attachedRepos ?? []).map((repo) => repo.dir), + ]); + for (const dir of dirs) { + if (!dir) continue; + for (const portal of await readHostPortalRegistry(dir)) { + if ( + !portal.sessionId || + !ownerIds.has(portal.sessionId) || + portal.state === "stopped" + ) + continue; + await stopPortalService({ + sessionId: session.id, + worktreeDir: dir, + name: portal.name, + expectedGeneration: portalGeneration(portal), + }); + } + } +} + +type ArchivedPortalOwner = Pick< + UnifiedSession, + "id" | "worktreeDir" | "attachedRepos" | "aliasIds" | "runner" | "sandbox" +>; + +/** + * The direct detail lookup answers a Slack or Linear id with that file's own + * row, alias and all, so archiving through an alias saw an owner id no Portal + * record carried. The merged list projection is the one place a historical + * alias resolves to the canonical session that absorbed it. + */ +async function findMergedSession( + sessionId: string, +): Promise { + const { findSessionAsync, getCachedSessionsAsync } = + await import("./session-cache"); + const merged = (await getCachedSessionsAsync()).find( + (session) => + session.id === sessionId || session.aliasIds?.includes(sessionId), + ); + // A session the list has not observed yet has no aliases to merge. + return merged ?? findSessionAsync(sessionId); +} + /** * A release predating Portal scopes can leave live preview trees inside the * gateway service cgroup. Restart only durable, live-owned Portal records so @@ -1181,22 +1495,25 @@ export async function migrateUnscopedPortalServices( ): Promise { if (!systemdUserScopesAvailable()) return { migrated: [] }; const owners = new Map>(); - const addOwner = (dir: string | null | undefined, sessionId: string) => { + const addOwner = async ( + dir: string | null | undefined, + session: PortalOwnerSession, + ) => { if (!dir) return; - const key = canonicalDir(dir); + const key = await canonicalDir(dir); const set = owners.get(key) ?? new Set(); - set.add(sessionId); + for (const id of portalOwnerIds(session)) set.add(id); owners.set(key, set); }; for (const session of sessions) { - addOwner(session.worktreeDir, session.id); + await addOwner(session.worktreeDir, session); for (const repo of session.attachedRepos ?? []) - addOwner(repo.dir, session.id); + await addOwner(repo.dir, session); } const migrated: PortalContainmentMigrationResult["migrated"] = []; for (const [worktreeDir, liveOwners] of owners) { - const records = readPortalRegistry(worktreeDir); + const records = await readHostPortalRegistry(worktreeDir); for (const portal of portalsNeedingContainment(records, true)) { const owned = portal.sessionId ? liveOwners.has(portal.sessionId) @@ -1233,53 +1550,42 @@ let portalReconcileInFlight = false; /** Reconcile Portal process groups after boot and every five minutes. */ export function startPortalReaper( - getSessions: () => readonly PortalOwnerSession[] = () => [], + getSessions: () => Promise, ): void { if (portalReapTimer) return; - const run = () => { + const run = async () => { if (portalReconcileInFlight) return; - let sessions: readonly PortalOwnerSession[]; + portalReconcileInFlight = true; try { - sessions = getSessions(); - } catch (error) { - console.error( - "[portals] session snapshot failed; skipping orphan reap:", - error, + const sessions = await getSessions(); + const { stopped } = await reapOrphanedPortalServices(sessions); + if (stopped.length) + console.log(`[portals] reaped ${stopped.length} Portal service(s)`); + const { slept } = await sleepIdlePortalServices(sessions); + if (slept.length) + console.log(`[portals] slept ${slept.length} idle Portal service(s)`); + const { migrated } = await migrateUnscopedPortalServices( + sessions.filter((session) => !session.archived), ); - return; + if (migrated.length) + console.log( + `[portals] migrated ${migrated.length} Portal service(s) into private scopes`, + ); + } catch (error) { + console.error("[portals] reconciliation failed:", error); + } finally { + portalReconcileInFlight = false; } - portalReconcileInFlight = true; - void reapOrphanedPortalServices(sessions) - .then(async ({ stopped }) => { - if (stopped.length) - console.log( - `[portals] reaped ${stopped.length} orphaned Portal service(s)`, - ); - const { slept } = await sleepIdlePortalServices(sessions); - if (slept.length) - console.log(`[portals] slept ${slept.length} idle Portal service(s)`); - const { migrated } = await migrateUnscopedPortalServices(sessions); - if (migrated.length) - console.log( - `[portals] migrated ${migrated.length} Portal service(s) into private scopes`, - ); - }) - .catch((error) => - console.error("[portals] reconciliation failed:", error), - ) - .finally(() => { - portalReconcileInFlight = false; - }); }; - run(); - portalReapTimer = setInterval(run, PORTAL_REAP_INTERVAL_MS); + void run(); + portalReapTimer = setInterval(() => void run(), PORTAL_REAP_INTERVAL_MS); portalReapTimer.unref?.(); console.log( - `[portals] orphan reaper started (every ${PORTAL_REAP_INTERVAL_MS / 60_000}m)`, + `[portals] lifecycle reaper started (every ${PORTAL_REAP_INTERVAL_MS / 60_000}m)`, ); } -export async function restartPortalService(input: { +export function restartPortalService(input: { sessionId: string; worktreeDir: string; name: string; @@ -1287,30 +1593,37 @@ export async function restartPortalService(input: { readyTimeoutMs?: number; }): Promise { const name = validateName(input.name); - const current = readPortalRegistry(input.worktreeDir).find( - (record) => record.name === name, - ); - if (!current) throw new Error(`Portal '${name}' does not exist.`); - await stopPortalService(input); - return startPortalService({ - ...input, - name, - key: current.key, - command: current.command, - port: current.port, - description: current.description, - defaultPath: current.defaultPath, - readyTimeoutMs: input.readyTimeoutMs ?? current.readyTimeoutMs, + // One transaction: nothing may stop or start this Portal between the stop + // of the old process and the registration of its replacement. + return withHostPortalOperation(input.worktreeDir, name, async () => { + const current = (await readHostPortalRegistry(input.worktreeDir)).find( + (record) => record.name === name, + ); + if (!current) throw new Error(`Portal '${name}' does not exist.`); + await stopHostPortal(input); + return startHostPortal({ + ...input, + name, + key: current.key, + command: current.command, + port: current.port, + description: current.description, + defaultPath: current.defaultPath, + readyTimeoutMs: input.readyTimeoutMs ?? current.readyTimeoutMs, + }); }); } -export function setPortalPath( +export async function setPortalPath( worktreeDir: string, path: string, name?: string, -): PortalRecord[] { - const next = withPortalPath(readPortalRegistry(worktreeDir), path, name); - writePortalRegistry(worktreeDir, next); +): Promise { + let next: PortalRecord[] = []; + await updateHostPortalRegistry(worktreeDir, (records) => { + next = withPortalPath(records, path, name); + return next; + }); return next; } diff --git a/packages/core/opensession-server/src/server/portals-mcp.ts b/packages/core/opensession-server/src/server/portals-mcp.ts index c79b9f9d2..e3fc04f3d 100644 --- a/packages/core/opensession-server/src/server/portals-mcp.ts +++ b/packages/core/opensession-server/src/server/portals-mcp.ts @@ -581,7 +581,7 @@ export function createPortalsMcpServer(ctx: PortalsMcpContext) { "Could not set Portal route: this session's Sandbox is sleeping or unavailable.", ); if (sandbox) await setSandboxPortalPath(sandbox, path, name); - else setPortalPath(dir, path, name); + else await setPortalPath(dir, path, name); } else { const normalized = normalizePortalPath(path) ?? null; await ctx.setDefaultPath(normalized); diff --git a/packages/core/opensession-server/src/server/routes/preview.ts b/packages/core/opensession-server/src/server/routes/preview.ts index 8ede247bd..aed44dc32 100644 --- a/packages/core/opensession-server/src/server/routes/preview.ts +++ b/packages/core/opensession-server/src/server/routes/preview.ts @@ -36,6 +36,7 @@ import { import { getRepo } from "../worktree"; import { configuredServer } from "../config"; import { portalNavigationRequest } from "../portal-sign-in"; +import { hostPortalActivity } from "../portal-lifecycle"; import { portalWaitingResponse } from "../portal-waiting-page"; import { sleepingSandboxPortalStatus } from "../sandbox-portals"; import type { UnifiedSession } from "../types"; @@ -98,7 +99,7 @@ export async function handlePreviewRoutes( try { // Host Portals keep their authenticated Caddy route while sleeping. A // real navigation wakes one; background fetches from stale tabs do not. - const hostPortal = hostPortalRouteStatus(httpsPort - 6_000); + const hostPortal = await hostPortalRouteStatus(httpsPort - 6_000); if (hostPortal) { if (!portalRouteAuthorized(httpsPort)) return notActive(hostPortal.sessionId); @@ -193,6 +194,9 @@ export async function handlePreviewRoutes( }, }); } + // Only authenticated, authorized Portal traffic counts. Session-list and + // readiness polling never extend a host preview's idle lifetime. + hostPortalActivity.touch(httpsPort - 6_000); return new Response(null, { status: 204, headers: { "Cache-Control": "no-store" },