From ab7a7a82b7d760832cbfbefbcb4a77395ff7ed4f Mon Sep 17 00:00:00 2001 From: Stack Test Date: Mon, 3 Aug 2026 08:58:27 +0200 Subject: [PATCH 1/3] feat(server): reap idle dev stacks from a level-triggered sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dev stacks that repositories start outside T3's process tree have been surviving for hours after the work that needed them finished. In the case that prompted this, ten stacks stayed up for thirteen hours across two already-merged PRs, holding roughly 15 GB of RSS. Teardown was never broken; nothing invoked it. Both existing triggers are edge-triggered. runOnWorktreeRemove needs a worktree removal, and merging a PR does not remove one. runOnPrMerged fires from VcsStatusBroadcaster only when it observes a not-merged -> merged transition for a worktree it happens to be polling, so an agent that finishes its work and a human who merges an hour later produce no observer, no edge, and no teardown. Add a sweep that looks at state rather than events, so a missed hook costs one idle window instead of an unbounded number of hours. Both project scripts stay: they are the early reap and are strictly faster when they do fire. This is the floor under them, not a replacement. The registry format (`dev-stack/1`, contracts/devStack.ts) carries facts only — pids, ports, and where each process runs — because the sweep reads every project's stacks and cannot hold per-repo knowledge. Policy is declared per checkout under `devStacks` in t3.json: idle window, consumer patterns, and the ordered entry roles. That split matters beyond tidiness: an idle window is a property of a repository rather than of one running instance, so changing it now applies to stacks that are already up, on the next sweep. Decisions worth knowing: - Opt-in. A repository without `devStacks` is left alone rather than swept under a guessed policy. - `entryRoles` is ordered and we watch the first role present. A frontend holds keep-alive connections to its own API, so also watching the API port would read a stack as busy for as long as the frontend is up, and nothing would ever look idle. An api-only stack falls through to the API by the same rule. - An absent `entryRoles` watches every port. That over-detects activity, which is the right way to be wrong: a stack lives too long instead of dying mid-run. - A stack is never reaped on the sweep that first sees it. A cold boot plus a build can outlast the window, so the clock starts from an observation. - The producer's start/stop lock is honoured, so a stack that is halfway up is skipped rather than killed. - Ownership is confirmed against /proc//cwd before signalling, so a recycled PID is never mistaken for a stack's own process. The sweep runs as one layer-scoped fiber in the server that already runs, rather than a separate unit: the server is what starts the agents that create these stacks, so "server down" means "no new stacks", and a per-stack systemd TTL is the better answer for a hung server than a second daemon would be. Failures are logged and swallowed. Each sweep reports its duration and counts, and logs at info when it acted or ran long, so whether it burdens the main loop stays a query rather than a guess. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/devStacks/DevStackPolicy.test.ts | 150 ++++++++ apps/server/src/devStacks/DevStackPolicy.ts | 103 ++++++ apps/server/src/devStacks/DevStackReaper.ts | 341 ++++++++++++++++++ apps/server/src/server.ts | 14 +- packages/contracts/src/devStack.ts | 59 +++ packages/contracts/src/index.ts | 1 + packages/contracts/src/t3ProjectFile.ts | 49 +++ 7 files changed, 716 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/devStacks/DevStackPolicy.test.ts create mode 100644 apps/server/src/devStacks/DevStackPolicy.ts create mode 100644 apps/server/src/devStacks/DevStackReaper.ts create mode 100644 packages/contracts/src/devStack.ts diff --git a/apps/server/src/devStacks/DevStackPolicy.test.ts b/apps/server/src/devStacks/DevStackPolicy.test.ts new file mode 100644 index 00000000000..467e1f3c50c --- /dev/null +++ b/apps/server/src/devStacks/DevStackPolicy.test.ts @@ -0,0 +1,150 @@ +import { DEV_STACK_SCHEMA_VERSION, type DevStackEntry } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import { + DEFAULT_CONSUMERS, + DEFAULT_IDLE_MINUTES, + decide, + entryPortsFor, + isActive, + resolvePolicy, +} from "./DevStackPolicy.ts"; + +const ROOT = "/var/lib/t3/worktrees/scanner/feature"; + +const entry = (processes: DevStackEntry["processes"]): DevStackEntry => ({ + schema: DEV_STACK_SCHEMA_VERSION, + project: "scanner", + worktree: "bdb56cccb27eb374", + root: ROOT, + instance: "empasa", + processes, +}); + +const browserStack = entry([ + { role: "api", pid: 1, port: 21000, cwd: "api" }, + { role: "frontend", pid: 2, port: 22000, cwd: "frontend" }, +]); +const apiOnlyStack = entry([{ role: "api", pid: 1, port: 21000, cwd: "api" }]); + +const idle = { establishedPorts: new Set(), consumerCwds: [] }; +const policy = resolvePolicy({ entryRoles: ["frontend", "api"] }); + +describe("resolvePolicy", () => { + it("applies the documented defaults when a repository declares nothing", () => { + expect(resolvePolicy(undefined)).toEqual({ + idleMs: DEFAULT_IDLE_MINUTES * 60_000, + consumers: DEFAULT_CONSUMERS, + entryRoles: [], + }); + }); + + it("takes the repository's declaration over the defaults", () => { + expect(resolvePolicy({ idleMinutes: 45, consumers: ["cypress"], entryRoles: ["web"] })).toEqual( + { + idleMs: 45 * 60_000, + consumers: ["cypress"], + entryRoles: ["web"], + }, + ); + }); +}); + +describe("entryPortsFor", () => { + it("watches the frontend of a browser stack, not the API behind it", () => { + expect(entryPortsFor(browserStack, ["frontend", "api"])).toEqual([22000]); + }); + + it("falls through to the API when no frontend is running", () => { + expect(entryPortsFor(apiOnlyStack, ["frontend", "api"])).toEqual([21000]); + }); + + it("watches every port when the repository declared no entry roles", () => { + expect(entryPortsFor(browserStack, [])).toEqual([21000, 22000]); + }); + + it("watches nothing when declared roles match no running process", () => { + expect(entryPortsFor(browserStack, ["worker"])).toEqual([]); + }); +}); + +describe("isActive", () => { + it("ignores a connection to the API of a browser stack", () => { + expect(isActive(browserStack, policy, { ...idle, establishedPorts: new Set([21000]) })).toBe( + false, + ); + }); + + it("counts a connection to the entry port", () => { + expect(isActive(browserStack, policy, { ...idle, establishedPorts: new Set([22000]) })).toBe( + true, + ); + }); + + it("counts a consumer working anywhere inside the worktree", () => { + expect(isActive(browserStack, policy, { ...idle, consumerCwds: [`${ROOT}/e2e`] })).toBe(true); + expect(isActive(browserStack, policy, { ...idle, consumerCwds: [ROOT] })).toBe(true); + }); + + it("does not mistake a sibling worktree sharing a path prefix for a consumer", () => { + expect(isActive(browserStack, policy, { ...idle, consumerCwds: [`${ROOT}-other/e2e`] })).toBe( + false, + ); + }); +}); + +describe("decide", () => { + const base = { + entry: browserStack, + policy, + observation: idle, + anyProcessAlive: true, + rootExists: true, + lastSeenMs: null as number | null, + now: 1_000_000_000, + }; + + it("prunes bookkeeping when every process is already gone", () => { + expect(decide({ ...base, anyProcessAlive: false })).toEqual({ + _tag: "Prune", + reason: "no-live-processes", + }); + }); + + it("prefers pruning over reaping when the worktree is gone but so are the processes", () => { + expect(decide({ ...base, anyProcessAlive: false, rootExists: false })._tag).toBe("Prune"); + }); + + it("reaps a live stack whose worktree was removed, whatever the clock says", () => { + expect(decide({ ...base, rootExists: false, lastSeenMs: base.now })).toEqual({ + _tag: "Reap", + reason: "worktree-removed", + }); + }); + + it("starts the clock the first time it sees a stack rather than reaping it", () => { + expect(decide(base)).toEqual({ _tag: "StartClock" }); + }); + + it("keeps a stack that is still inside its idle window", () => { + const lastSeenMs = base.now - (policy.idleMs - 60_000); + expect(decide({ ...base, lastSeenMs })).toEqual({ + _tag: "Keep", + idleMs: policy.idleMs - 60_000, + }); + }); + + it("reaps once the idle window is exceeded", () => { + const lastSeenMs = base.now - (policy.idleMs + 60_000); + expect(decide({ ...base, lastSeenMs })).toEqual({ + _tag: "Reap", + reason: "idle", + idleMs: policy.idleMs + 60_000, + }); + }); + + it("refreshes an active stack instead of ageing it out", () => { + const observation = { ...idle, establishedPorts: new Set([22000]) }; + const lastSeenMs = base.now - (policy.idleMs + 60_000); + expect(decide({ ...base, observation, lastSeenMs })).toEqual({ _tag: "Active" }); + }); +}); diff --git a/apps/server/src/devStacks/DevStackPolicy.ts b/apps/server/src/devStacks/DevStackPolicy.ts new file mode 100644 index 00000000000..058e03454e8 --- /dev/null +++ b/apps/server/src/devStacks/DevStackPolicy.ts @@ -0,0 +1,103 @@ +import type { DevStackEntry, T3ProjectFileDevStacks } from "@t3tools/contracts"; +import * as path from "node:path"; + +/** + * The decision half of the dev stack sweep, kept free of the filesystem, /proc, + * and signals so it can be tested directly rather than through a live stack. + */ + +export const DEFAULT_IDLE_MINUTES = 20; +export const DEFAULT_CONSUMERS: ReadonlyArray = Object.freeze(["playwright", "vitest"]); + +export interface DevStackPolicy { + readonly idleMs: number; + readonly consumers: ReadonlyArray; + readonly entryRoles: ReadonlyArray; +} + +/** + * `entryRoles` has no safe generic default, so an absent one means "watch every + * port in the stack". That over-detects activity, which is the right way to be + * wrong: a repository that has not declared its entry role keeps a stack alive + * too long rather than losing one mid-run. + */ +export const resolvePolicy = (declared: T3ProjectFileDevStacks | undefined): DevStackPolicy => ({ + idleMs: (declared?.idleMinutes ?? DEFAULT_IDLE_MINUTES) * 60_000, + consumers: declared?.consumers ?? DEFAULT_CONSUMERS, + entryRoles: declared?.entryRoles ?? [], +}); + +/** + * Ports a consumer would connect to. With `entryRoles` declared we watch the + * first role actually present, most specific first — a frontend usually holds + * keep-alive connections to its own API, so watching the API port too would read + * the stack as busy for as long as the frontend is up. + */ +export const entryPortsFor = ( + entry: DevStackEntry, + entryRoles: ReadonlyArray, +): ReadonlyArray => { + const portsOf = (role: string) => + entry.processes + .filter((process) => process.role === role && process.port !== undefined) + .map((process) => process.port as number); + for (const role of entryRoles) { + const ports = portsOf(role); + if (ports.length > 0) return ports; + } + if (entryRoles.length > 0) return []; + return entry.processes.flatMap((process) => (process.port === undefined ? [] : [process.port])); +}; + +const within = (child: string, parent: string) => + child === parent || child.startsWith(`${parent}${path.sep}`); + +export interface ActivityObservation { + readonly establishedPorts: ReadonlySet; + /** Working directories of live processes whose command line matched a consumer pattern. */ + readonly consumerCwds: ReadonlyArray; +} + +export const isActive = ( + entry: DevStackEntry, + policy: DevStackPolicy, + observation: ActivityObservation, +): boolean => { + for (const port of entryPortsFor(entry, policy.entryRoles)) { + if (observation.establishedPorts.has(port)) return true; + } + return observation.consumerCwds.some((cwd) => within(cwd, entry.root)); +}; + +export type SweepDecision = + | { readonly _tag: "Active" } + /** First sighting starts the clock — a cold boot plus a build can outlast the window. */ + | { readonly _tag: "StartClock" } + | { readonly _tag: "Keep"; readonly idleMs: number } + | { readonly _tag: "Reap"; readonly reason: "idle"; readonly idleMs: number } + | { readonly _tag: "Reap"; readonly reason: "worktree-removed" } + | { readonly _tag: "Prune"; readonly reason: "no-live-processes" }; + +export interface SweepInput { + readonly entry: DevStackEntry; + readonly policy: DevStackPolicy; + readonly observation: ActivityObservation; + readonly anyProcessAlive: boolean; + readonly rootExists: boolean; + /** mtime of the sibling `.seen` marker, or null when the sweep has never seen this stack. */ + readonly lastSeenMs: number | null; + readonly now: number; +} + +export const decide = (input: SweepInput): SweepDecision => { + // Nothing left to signal: a reboot, an OOM, or a manual kill already took the + // processes and only the bookkeeping survived. + if (!input.anyProcessAlive) return { _tag: "Prune", reason: "no-live-processes" }; + if (!input.rootExists) return { _tag: "Reap", reason: "worktree-removed" }; + if (isActive(input.entry, input.policy, input.observation)) return { _tag: "Active" }; + if (input.lastSeenMs === null) return { _tag: "StartClock" }; + const idleMs = input.now - input.lastSeenMs; + return idleMs > input.policy.idleMs + ? { _tag: "Reap", reason: "idle", idleMs } + : { _tag: "Keep", idleMs }; +}; diff --git a/apps/server/src/devStacks/DevStackReaper.ts b/apps/server/src/devStacks/DevStackReaper.ts new file mode 100644 index 00000000000..ec7796133fa --- /dev/null +++ b/apps/server/src/devStacks/DevStackReaper.ts @@ -0,0 +1,341 @@ +import { DevStackEntry, DEV_STACK_REGISTRY_DIR } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import * as T3ProjectFileLoader from "../project/T3ProjectFileLoader.ts"; +import { + decide, + resolvePolicy, + type ActivityObservation, + type DevStackPolicy, +} from "./DevStackPolicy.ts"; + +/** + * Stops dev stacks a repository registered under `$TMPDIR/dev-stacks` once + * nothing has used them for the window that repository declared in `t3.json`. + * + * This is the level-triggered half of stack teardown. The `runOnWorktreeRemove` + * and `runOnPrMerged` project scripts remain the early reap and are strictly + * faster when they fire; they simply cannot be relied on, because both are + * edge-triggered. `runOnPrMerged` in particular only fires when + * VcsStatusBroadcaster observes a not-merged -> merged transition for a worktree + * it happens to be polling, so an agent that finishes its work and a human who + * merges an hour later produce no observer, no edge, and no teardown. Stacks have + * survived for thirteen hours that way. + * + * Looking at state instead of events means a missed hook costs one idle window + * rather than an unbounded number of hours, and it needs no cooperation from the + * repository beyond registering the stack in the first place. + */ + +const SWEEP_INTERVAL = Duration.minutes(5); + +const decodeEntry = Schema.decodeUnknownOption(DevStackEntry); + +export interface SweepSummary { + readonly scanned: number; + readonly reaped: number; + readonly pruned: number; + readonly active: number; + readonly skipped: number; + readonly durationMs: number; +} + +export class DevStackReaper extends Context.Service< + DevStackReaper, + { + /** Run one sweep now. Exposed so a caller can force one; the fiber drives the rest. */ + readonly sweep: () => Effect.Effect; + } +>()("t3/devStacks/DevStackReaper") {} + +const registryRoot = () => path.join(os.tmpdir(), DEV_STACK_REGISTRY_DIR); + +const seenPathFor = (stackFile: string) => stackFile.replace(/\.json$/u, ".seen"); + +const listFiles = async (dir: string, suffix: string): Promise> => { + const out: string[] = []; + const projects = await fs.readdir(dir, { withFileTypes: true }).catch(() => []); + for (const project of projects) { + if (!project.isDirectory()) continue; + const projectDir = path.join(dir, project.name); + const worktrees = await fs.readdir(projectDir, { withFileTypes: true }).catch(() => []); + for (const worktree of worktrees) { + if (!worktree.isDirectory()) continue; + const worktreeDir = path.join(projectDir, worktree.name); + const entries = await fs.readdir(worktreeDir).catch(() => []); + for (const name of entries) if (name.endsWith(suffix)) out.push(path.join(worktreeDir, name)); + } + } + return out; +}; + +const alive = (pid: number) => { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +}; + +/** Confirms a PID is still the stack's own process before it is signalled. */ +const ownedBy = async (pid: number, expectedCwd: string) => { + if (!alive(pid)) return false; + try { + const [actual, expected] = await Promise.all([ + fs.realpath(`/proc/${pid}/cwd`), + fs.realpath(expectedCwd), + ]); + return actual === expected; + } catch { + return false; + } +}; + +/** + * Local ports with an ESTABLISHED connection, read from /proc so the sweep does + * not shell out. Field 1 is `HEXIP:HEXPORT`, field 3 is the state (01 = ESTABLISHED). + */ +const establishedLocalPorts = async (): Promise> => { + const ports = new Set(); + for (const file of ["/proc/net/tcp", "/proc/net/tcp6"]) { + const text = await fs.readFile(file, "utf8").catch(() => ""); + for (const line of text.split("\n").slice(1)) { + const fields = line.trim().split(/\s+/u); + if (fields.length < 4 || fields[3] !== "01") continue; + const hexPort = fields[1]?.split(":")[1]; + if (hexPort === undefined) continue; + const port = Number.parseInt(hexPort, 16); + if (Number.isFinite(port)) ports.add(port); + } + } + return ports; +}; + +/** Working directories of live processes whose command line matches any consumer pattern. */ +const consumerCwds = async (patterns: ReadonlySet): Promise> => { + if (patterns.size === 0) return []; + const cwds: string[] = []; + const entries = await fs.readdir("/proc").catch(() => []); + for (const entry of entries) { + if (!/^\d+$/u.test(entry)) continue; + try { + const cmdline = (await fs.readFile(`/proc/${entry}/cmdline`, "utf8")).replaceAll("\0", " "); + let matched = false; + for (const pattern of patterns) { + if (cmdline.includes(pattern)) { + matched = true; + break; + } + } + if (!matched) continue; + cwds.push(await fs.realpath(`/proc/${entry}/cwd`)); + } catch { + // Exited mid-scan, or owned by another user. Either way it is not ours. + } + } + return cwds; +}; + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** SIGTERM the process group, then SIGKILL what is left. */ +const stopGroup = async (pid: number) => { + try { + process.kill(-pid, "SIGTERM"); + } catch { + return; + } + for (let attempt = 0; attempt < 20 && alive(pid); attempt++) await wait(100); + if (!alive(pid)) return; + try { + process.kill(-pid, "SIGKILL"); + } catch { + // Exited between the check and the signal. + } +}; + +const releaseLease = async (base: string, pid: number, port: number | undefined) => { + if (port === undefined) return; + const file = path.join(base, "leases", `${port}.json`); + try { + const lease: unknown = JSON.parse(await fs.readFile(file, "utf8")); + if ((lease as { pid?: number }).pid !== pid) return; + } catch { + return; + } + await fs.rm(file, { force: true }); +}; + +/** + * The producer takes this lock around start and stop. Honouring it keeps the + * sweep from killing a stack that is halfway through coming up, and costs one + * stat per stack. + */ +const isLocked = async (base: string, entry: DevStackEntry) => { + const lock = path.join(base, "locks", `${entry.project}-${entry.worktree}-${entry.instance}`); + try { + const owner: unknown = JSON.parse(await fs.readFile(path.join(lock, "owner.json"), "utf8")); + const pid = (owner as { pid?: number }).pid; + return typeof pid === "number" && alive(pid); + } catch { + return false; + } +}; + +/** Reverse start order, so a frontend stops before the API it points at. */ +const teardown = async (base: string, entry: DevStackEntry, stackFile: string) => { + for (const process_ of [...entry.processes].reverse()) { + if (await ownedBy(process_.pid, path.join(entry.root, process_.cwd))) + await stopGroup(process_.pid); + await releaseLease(base, process_.pid, process_.port); + } + await fs.rm(stackFile, { force: true }); + await fs.rm(seenPathFor(stackFile), { force: true }); +}; + +export const make = Effect.gen(function* DevStackReaperMake() { + const projectFiles = yield* T3ProjectFileLoader.T3ProjectFileLoader; + + const policyFor = Effect.fn("DevStackReaper.policyFor")(function* (root: string) { + const file = yield* projectFiles.load(root); + return Option.match(file, { + onNone: () => null, + onSome: (loaded) => (loaded.devStacks === undefined ? null : resolvePolicy(loaded.devStacks)), + }); + }); + + const sweep = Effect.fn("DevStackReaper.sweep")(function* () { + const startedAt = Date.now(); + const base = registryRoot(); + const stackFiles = yield* Effect.promise(() => listFiles(path.join(base, "stacks"), ".json")); + const summary = { scanned: 0, reaped: 0, pruned: 0, active: 0, skipped: 0 }; + if (stackFiles.length === 0) return { ...summary, durationMs: Date.now() - startedAt }; + + const establishedPorts = yield* Effect.promise(establishedLocalPorts); + // Policies are resolved first so one /proc scan can serve every stack. + const resolved: Array<{ file: string; entry: DevStackEntry; policy: DevStackPolicy }> = []; + for (const file of stackFiles) { + summary.scanned++; + const raw = yield* Effect.promise(() => + fs + .readFile(file, "utf8") + .then(JSON.parse) + .catch(() => null), + ); + const decoded = raw === null ? Option.none() : decodeEntry(raw); + if (Option.isNone(decoded)) { + // Not a shape we understand. Only bookkeeping is removed; no signals sent. + yield* Effect.promise(() => fs.rm(file, { force: true })); + yield* Effect.promise(() => fs.rm(seenPathFor(file), { force: true })); + summary.pruned++; + continue; + } + const entry = decoded.value; + // A repository opts in by declaring devStacks. Without it, T3 leaves the + // stack alone entirely rather than guessing a policy for someone else's repo. + const policy = yield* policyFor(entry.root); + if (policy === null) { + summary.skipped++; + continue; + } + resolved.push({ file, entry, policy }); + } + if (resolved.length === 0) return { ...summary, durationMs: Date.now() - startedAt }; + + const patterns = new Set(resolved.flatMap(({ policy }) => [...policy.consumers])); + const cwds = yield* Effect.promise(() => consumerCwds(patterns)); + const observation: ActivityObservation = { establishedPorts, consumerCwds: cwds }; + + for (const { file, entry, policy } of resolved) { + if (yield* Effect.promise(() => isLocked(base, entry))) { + summary.skipped++; + continue; + } + const anyProcessAlive = entry.processes.some((process_) => alive(process_.pid)); + const rootExists = yield* Effect.promise(() => + fs + .stat(entry.root) + .then(() => true) + .catch(() => false), + ); + const lastSeenMs = yield* Effect.promise(() => + fs + .stat(seenPathFor(file)) + .then((stat) => stat.mtimeMs) + .catch(() => null), + ); + const decision = decide({ + entry, + policy, + observation, + anyProcessAlive, + rootExists, + lastSeenMs, + now: Date.now(), + }); + + switch (decision._tag) { + case "Active": + case "StartClock": { + const when = new Date(); + yield* Effect.promise(async () => { + await fs.writeFile(seenPathFor(file), "").catch(() => {}); + await fs.utimes(seenPathFor(file), when, when).catch(() => {}); + }); + summary.active++; + break; + } + case "Keep": + summary.active++; + break; + case "Prune": + yield* Effect.promise(() => teardown(base, entry, file)); + summary.pruned++; + break; + case "Reap": + yield* Effect.promise(() => teardown(base, entry, file)); + summary.reaped++; + yield* Effect.logInfo("dev stack reaped").pipe( + Effect.annotateLogs({ + project: entry.project, + instance: entry.instance, + root: entry.root, + reason: decision.reason, + }), + ); + break; + } + } + + return { ...summary, durationMs: Date.now() - startedAt }; + }); + + // One layer-scoped fiber. Failures are logged and swallowed: a sweep that + // cannot read /proc must not take the server down with it. + const tick = () => + sweep().pipe( + Effect.flatMap((summary) => + summary.reaped + summary.pruned > 0 || summary.durationMs > 250 + ? // Routine quiet sweeps stay at debug; anything that acted or ran long is + // worth seeing, so "is this bogging the main loop down" stays a query. + Effect.logInfo("dev stack sweep").pipe(Effect.annotateLogs({ ...summary })) + : Effect.logDebug("dev stack sweep").pipe(Effect.annotateLogs({ ...summary })), + ), + Effect.catchCause((cause) => Effect.logWarning("dev stack sweep failed", cause)), + ); + + yield* Effect.forkScoped(tick().pipe(Effect.repeat(Schedule.spaced(SWEEP_INTERVAL)))); + + return { sweep } as const; +}); + +export const layer = Layer.effect(DevStackReaper, make); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 03c1444fb02..3af423b5d39 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -62,6 +62,7 @@ import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as ProjectFaviconResolver from "./project/ProjectFaviconResolver.ts"; +import * as DevStackReaper from "./devStacks/DevStackReaper.ts"; import * as T3ProjectFileLoader from "./project/T3ProjectFileLoader.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts"; @@ -379,6 +380,10 @@ const ProjectFaviconResolverLayerLive = ProjectFaviconResolver.layer.pipe( Layer.provide(T3ProjectFileLoader.layer), ); +// Self-contained: the sweep reads the dev-stack registry and /proc directly, and +// only needs t3.json to learn each repository's declared policy. +const DevStackReaperLayerLive = DevStackReaper.layer.pipe(Layer.provide(T3ProjectFileLoader.layer)); + const AuthLayerLive = EnvironmentAuth.layer.pipe( Layer.provideMerge(PersistenceLayerLive), Layer.provide(ServerSecretStore.layer), @@ -417,7 +422,14 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(GitLayerLive), Layer.provideMerge(VcsLayerLive), Layer.provideMerge(ProviderRuntimeLayerLive), - Layer.provideMerge(Layer.mergeAll(TerminalLayerLive, PreviewLayerLive, AiUsageMonitor.layer)), + Layer.provideMerge( + Layer.mergeAll( + TerminalLayerLive, + PreviewLayerLive, + AiUsageMonitor.layer, + DevStackReaperLayerLive, + ), + ), Layer.provideMerge(PersistenceLayerLive), Layer.provideMerge(Keybindings.layer), Layer.provideMerge(ProviderRegistryLive), diff --git a/packages/contracts/src/devStack.ts b/packages/contracts/src/devStack.ts new file mode 100644 index 00000000000..a989331a6f9 --- /dev/null +++ b/packages/contracts/src/devStack.ts @@ -0,0 +1,59 @@ +import * as Schema from "effect/Schema"; + +/** + * `dev-stack/1` — the on-disk format a repository uses to register long-running + * dev servers it started outside T3's process tree, so T3 can supervise them. + * + * Entries live under `/stacks///.json`, + * where the registry defaults to `$TMPDIR/dev-stacks`. T3 sweeps the whole + * registry across every project, so an entry carries **facts only** — pids, + * ports, and where each process runs. Anything that requires judgement (how long + * idle is too long, what counts as a consumer, which role a consumer connects to) + * is declared per repository under `devStacks` in `t3.json`, because it is a + * property of the repository rather than of one running instance. + * + * The producer side of this contract currently lives in macs-scanner's + * `scripts/lib/agent-e2e-stack.mjs`. + */ +export const DEV_STACK_SCHEMA_VERSION = "dev-stack/1"; + +/** Default registry directory name, resolved inside the OS temp directory. */ +export const DEV_STACK_REGISTRY_DIR = "dev-stacks"; + +export const DevStackProcess = Schema.Struct({ + role: Schema.String.annotate({ + description: + 'Role within the stack, e.g. "api" or "frontend". Matched against `devStacks.entryRoles`.', + }), + pid: Schema.Int.check(Schema.isGreaterThan(0)).annotate({ + description: "PID of the process group leader. The stack is torn down by signalling the group.", + }), + port: Schema.optionalKey( + Schema.Int.check(Schema.isGreaterThan(0)).annotate({ + description: "Port the process listens on, verified by the producer's health check at start.", + }), + ), + cwd: Schema.String.annotate({ + description: + "Working directory relative to `root`. T3 confirms /proc//cwd matches before signalling, so a recycled PID is never mistaken for the stack's own process.", + }), +}).annotate({ description: "One process belonging to a registered dev stack." }); +export type DevStackProcess = typeof DevStackProcess.Type; + +export const DevStackEntry = Schema.Struct({ + schema: Schema.Literal(DEV_STACK_SCHEMA_VERSION), + project: Schema.String, + worktree: Schema.String.annotate({ description: "Stable hash of the worktree path." }), + root: Schema.String.annotate({ + description: "Absolute path of the worktree that owns the stack.", + }), + instance: Schema.String.annotate({ + description: + "What the repository shards stacks by — a company, a tenant, a variant. Opaque to T3; it only has to be unique within a worktree.", + }), + processes: Schema.Array(DevStackProcess), +}).annotate({ + title: "Dev stack registry entry", + description: "A live dev stack registered for T3 to supervise.", +}); +export type DevStackEntry = typeof DevStackEntry.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index d7064a7cf62..107615874d0 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -19,6 +19,7 @@ export * from "./vcs.ts"; export * from "./sourceControl.ts"; export * from "./orchestration.ts"; export * from "./t3ProjectFile.ts"; +export * from "./devStack.ts"; export * from "./editor.ts"; export * from "./openWith.ts"; export * from "./project.ts"; diff --git a/packages/contracts/src/t3ProjectFile.ts b/packages/contracts/src/t3ProjectFile.ts index 4fcb0950d7a..cd33b23c5a0 100644 --- a/packages/contracts/src/t3ProjectFile.ts +++ b/packages/contracts/src/t3ProjectFile.ts @@ -11,6 +11,8 @@ export const T3_PROJECT_FILE_SCHEMA_URL = "https://t3.codes/schema/t3.json"; const T3_PROJECT_FILE_PATH_MAX_LENGTH = 512; const T3_PROJECT_FILE_MAX_SCRIPTS = 50; +const T3_PROJECT_FILE_MAX_DEV_STACK_CONSUMERS = 20; +const T3_PROJECT_FILE_MAX_DEV_STACK_ROLES = 10; // Annotations go on the encoded (string) side so they survive into the // published JSON Schema; decoding still trims and re-validates non-emptiness. @@ -70,6 +72,47 @@ export const T3ProjectFileScript = Schema.Struct({ }); export type T3ProjectFileScript = typeof T3ProjectFileScript.Type; +/** + * Policy for long-running dev servers a repository starts outside T3's own + * process tree (test stacks, preview servers). The live instances themselves are + * registered under `$TMPDIR/dev-stacks` in the repo-agnostic `dev-stack/1` format; + * this declares how T3 should treat them. + * + * Policy lives here rather than in each registry entry on purpose: an idle window + * is a property of the repository, not of one running instance, and changing it + * should apply to instances that are already up on the next sweep instead of only + * to ones started afterwards. + */ +export const T3ProjectFileDevStacks = Schema.Struct({ + idleMinutes: Schema.optionalKey( + Schema.Number.check(Schema.isGreaterThan(0)).annotate({ + description: + "Minutes a registered dev stack may go without an observed consumer before T3 stops it. Defaults to 20.", + }), + ), + consumers: Schema.optionalKey( + Schema.Array( + trimmedNonEmpty({ description: "Substring matched against process command lines." }), + ) + .annotate({ + description: + 'Processes whose presence inside the worktree counts as the stack being in use, e.g. ["playwright", "vitest"]. Deliberately narrow: a shell sitting in the worktree is not a consumer, and treating it as one would keep every stack alive for the length of a session.', + }) + .check(Schema.isMaxLength(T3_PROJECT_FILE_MAX_DEV_STACK_CONSUMERS)), + ), + entryRoles: Schema.optionalKey( + Schema.Array(trimmedNonEmpty({ description: "A role name used in the dev-stack registry." })) + .annotate({ + description: + 'Roles a consumer connects to, most specific first, e.g. ["frontend", "api"]. T3 watches the first role present in an instance. Ordering matters: a frontend usually holds keep-alive connections to its own API, so counting the API port would read the stack as busy for as long as the frontend is up.', + }) + .check(Schema.isMaxLength(T3_PROJECT_FILE_MAX_DEV_STACK_ROLES)), + ), +}).annotate({ + description: "How T3 supervises long-running dev stacks this repository registers.", +}); +export type T3ProjectFileDevStacks = typeof T3ProjectFileDevStacks.Type; + export const T3ProjectFile = Schema.Struct({ $schema: Schema.optionalKey( Schema.String.annotate({ @@ -92,6 +135,12 @@ export const T3ProjectFile = Schema.Struct({ }) .check(Schema.isMaxLength(T3_PROJECT_FILE_MAX_SCRIPTS)), ), + devStacks: Schema.optionalKey( + T3ProjectFileDevStacks.annotate({ + description: + "Opt in to T3 supervising the dev stacks this repository registers under $TMPDIR/dev-stacks. Omit it and T3 leaves them alone.", + }), + ), }).annotate({ title: "T3 project file", description: From 0ed7ae186cc42b292ec6e220c6105a4b67d0ce1a Mon Sep 17 00:00:00 2001 From: Stack Test Date: Mon, 3 Aug 2026 09:05:02 +0200 Subject: [PATCH 2/3] chore(contracts): keep the dev-stack contract free of downstream references The producer of dev-stack/1 is a private repository; naming it here, and using its worktree paths and tenant names as test fixtures, leaks downstream detail into a contract that is meant to be repo-agnostic and is read by anyone adopting the format. Co-Authored-By: Claude Opus 5 (1M context) --- apps/server/src/devStacks/DevStackPolicy.test.ts | 6 +++--- packages/contracts/src/devStack.ts | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/apps/server/src/devStacks/DevStackPolicy.test.ts b/apps/server/src/devStacks/DevStackPolicy.test.ts index 467e1f3c50c..cdd9723d19a 100644 --- a/apps/server/src/devStacks/DevStackPolicy.test.ts +++ b/apps/server/src/devStacks/DevStackPolicy.test.ts @@ -9,14 +9,14 @@ import { resolvePolicy, } from "./DevStackPolicy.ts"; -const ROOT = "/var/lib/t3/worktrees/scanner/feature"; +const ROOT = "/workspaces/example/feature"; const entry = (processes: DevStackEntry["processes"]): DevStackEntry => ({ schema: DEV_STACK_SCHEMA_VERSION, - project: "scanner", + project: "example", worktree: "bdb56cccb27eb374", root: ROOT, - instance: "empasa", + instance: "primary", processes, }); diff --git a/packages/contracts/src/devStack.ts b/packages/contracts/src/devStack.ts index a989331a6f9..949de06beb7 100644 --- a/packages/contracts/src/devStack.ts +++ b/packages/contracts/src/devStack.ts @@ -12,8 +12,8 @@ import * as Schema from "effect/Schema"; * is declared per repository under `devStacks` in `t3.json`, because it is a * property of the repository rather than of one running instance. * - * The producer side of this contract currently lives in macs-scanner's - * `scripts/lib/agent-e2e-stack.mjs`. + * Repositories write these entries themselves, from whatever starts the servers. + * T3 only reads them. */ export const DEV_STACK_SCHEMA_VERSION = "dev-stack/1"; From e63d6f60955af6c23f59b6b4f049598c8ff84b6d Mon Sep 17 00:00:00 2001 From: "omegent-app[bot]" <306514130+omegent-app[bot]@users.noreply.github.com> Date: Mon, 3 Aug 2026 11:23:15 +0000 Subject: [PATCH 3/3] fix(server): green dev stack reaper PR --- apps/server/src/devStacks/DevStackPolicy.ts | 5 +- apps/server/src/devStacks/DevStackReaper.ts | 76 +++++++++++---------- packages/shared/src/t3ProjectFile.test.ts | 14 +++- 3 files changed, 57 insertions(+), 38 deletions(-) diff --git a/apps/server/src/devStacks/DevStackPolicy.ts b/apps/server/src/devStacks/DevStackPolicy.ts index 058e03454e8..3981f96dc55 100644 --- a/apps/server/src/devStacks/DevStackPolicy.ts +++ b/apps/server/src/devStacks/DevStackPolicy.ts @@ -1,5 +1,6 @@ +// @effect-diagnostics nodeBuiltinImport:off import type { DevStackEntry, T3ProjectFileDevStacks } from "@t3tools/contracts"; -import * as path from "node:path"; +import * as NodePath from "node:path"; /** * The decision half of the dev stack sweep, kept free of the filesystem, /proc, @@ -50,7 +51,7 @@ export const entryPortsFor = ( }; const within = (child: string, parent: string) => - child === parent || child.startsWith(`${parent}${path.sep}`); + child === parent || child.startsWith(`${parent}${NodePath.sep}`); export interface ActivityObservation { readonly establishedPorts: ReadonlySet; diff --git a/apps/server/src/devStacks/DevStackReaper.ts b/apps/server/src/devStacks/DevStackReaper.ts index ec7796133fa..ada21493f69 100644 --- a/apps/server/src/devStacks/DevStackReaper.ts +++ b/apps/server/src/devStacks/DevStackReaper.ts @@ -1,3 +1,4 @@ +// @effect-diagnostics nodeBuiltinImport:off globalTimers:off globalDateInEffect:off import { DevStackEntry, DEV_STACK_REGISTRY_DIR } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Duration from "effect/Duration"; @@ -6,9 +7,9 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; -import * as fs from "node:fs/promises"; -import * as os from "node:os"; -import * as path from "node:path"; +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; import * as T3ProjectFileLoader from "../project/T3ProjectFileLoader.ts"; import { decide, @@ -56,22 +57,23 @@ export class DevStackReaper extends Context.Service< } >()("t3/devStacks/DevStackReaper") {} -const registryRoot = () => path.join(os.tmpdir(), DEV_STACK_REGISTRY_DIR); +const registryRoot = () => NodePath.join(NodeOS.tmpdir(), DEV_STACK_REGISTRY_DIR); const seenPathFor = (stackFile: string) => stackFile.replace(/\.json$/u, ".seen"); const listFiles = async (dir: string, suffix: string): Promise> => { const out: string[] = []; - const projects = await fs.readdir(dir, { withFileTypes: true }).catch(() => []); + const projects = await NodeFSP.readdir(dir, { withFileTypes: true }).catch(() => []); for (const project of projects) { if (!project.isDirectory()) continue; - const projectDir = path.join(dir, project.name); - const worktrees = await fs.readdir(projectDir, { withFileTypes: true }).catch(() => []); + const projectDir = NodePath.join(dir, project.name); + const worktrees = await NodeFSP.readdir(projectDir, { withFileTypes: true }).catch(() => []); for (const worktree of worktrees) { if (!worktree.isDirectory()) continue; - const worktreeDir = path.join(projectDir, worktree.name); - const entries = await fs.readdir(worktreeDir).catch(() => []); - for (const name of entries) if (name.endsWith(suffix)) out.push(path.join(worktreeDir, name)); + const worktreeDir = NodePath.join(projectDir, worktree.name); + const entries = await NodeFSP.readdir(worktreeDir).catch(() => []); + for (const name of entries) + if (name.endsWith(suffix)) out.push(NodePath.join(worktreeDir, name)); } } return out; @@ -91,8 +93,8 @@ const ownedBy = async (pid: number, expectedCwd: string) => { if (!alive(pid)) return false; try { const [actual, expected] = await Promise.all([ - fs.realpath(`/proc/${pid}/cwd`), - fs.realpath(expectedCwd), + NodeFSP.realpath(`/proc/${pid}/cwd`), + NodeFSP.realpath(expectedCwd), ]); return actual === expected; } catch { @@ -107,7 +109,7 @@ const ownedBy = async (pid: number, expectedCwd: string) => { const establishedLocalPorts = async (): Promise> => { const ports = new Set(); for (const file of ["/proc/net/tcp", "/proc/net/tcp6"]) { - const text = await fs.readFile(file, "utf8").catch(() => ""); + const text = await NodeFSP.readFile(file, "utf8").catch(() => ""); for (const line of text.split("\n").slice(1)) { const fields = line.trim().split(/\s+/u); if (fields.length < 4 || fields[3] !== "01") continue; @@ -124,11 +126,14 @@ const establishedLocalPorts = async (): Promise> => { const consumerCwds = async (patterns: ReadonlySet): Promise> => { if (patterns.size === 0) return []; const cwds: string[] = []; - const entries = await fs.readdir("/proc").catch(() => []); + const entries = await NodeFSP.readdir("/proc").catch(() => []); for (const entry of entries) { if (!/^\d+$/u.test(entry)) continue; try { - const cmdline = (await fs.readFile(`/proc/${entry}/cmdline`, "utf8")).replaceAll("\0", " "); + const cmdline = (await NodeFSP.readFile(`/proc/${entry}/cmdline`, "utf8")).replaceAll( + "\0", + " ", + ); let matched = false; for (const pattern of patterns) { if (cmdline.includes(pattern)) { @@ -137,7 +142,7 @@ const consumerCwds = async (patterns: ReadonlySet): Promise { const releaseLease = async (base: string, pid: number, port: number | undefined) => { if (port === undefined) return; - const file = path.join(base, "leases", `${port}.json`); + const file = NodePath.join(base, "leases", `${port}.json`); try { - const lease: unknown = JSON.parse(await fs.readFile(file, "utf8")); + const lease: unknown = JSON.parse(await NodeFSP.readFile(file, "utf8")); if ((lease as { pid?: number }).pid !== pid) return; } catch { return; } - await fs.rm(file, { force: true }); + await NodeFSP.rm(file, { force: true }); }; /** @@ -181,9 +186,11 @@ const releaseLease = async (base: string, pid: number, port: number | undefined) * stat per stack. */ const isLocked = async (base: string, entry: DevStackEntry) => { - const lock = path.join(base, "locks", `${entry.project}-${entry.worktree}-${entry.instance}`); + const lock = NodePath.join(base, "locks", `${entry.project}-${entry.worktree}-${entry.instance}`); try { - const owner: unknown = JSON.parse(await fs.readFile(path.join(lock, "owner.json"), "utf8")); + const owner: unknown = JSON.parse( + await NodeFSP.readFile(NodePath.join(lock, "owner.json"), "utf8"), + ); const pid = (owner as { pid?: number }).pid; return typeof pid === "number" && alive(pid); } catch { @@ -194,12 +201,12 @@ const isLocked = async (base: string, entry: DevStackEntry) => { /** Reverse start order, so a frontend stops before the API it points at. */ const teardown = async (base: string, entry: DevStackEntry, stackFile: string) => { for (const process_ of [...entry.processes].reverse()) { - if (await ownedBy(process_.pid, path.join(entry.root, process_.cwd))) + if (await ownedBy(process_.pid, NodePath.join(entry.root, process_.cwd))) await stopGroup(process_.pid); await releaseLease(base, process_.pid, process_.port); } - await fs.rm(stackFile, { force: true }); - await fs.rm(seenPathFor(stackFile), { force: true }); + await NodeFSP.rm(stackFile, { force: true }); + await NodeFSP.rm(seenPathFor(stackFile), { force: true }); }; export const make = Effect.gen(function* DevStackReaperMake() { @@ -216,7 +223,9 @@ export const make = Effect.gen(function* DevStackReaperMake() { const sweep = Effect.fn("DevStackReaper.sweep")(function* () { const startedAt = Date.now(); const base = registryRoot(); - const stackFiles = yield* Effect.promise(() => listFiles(path.join(base, "stacks"), ".json")); + const stackFiles = yield* Effect.promise(() => + listFiles(NodePath.join(base, "stacks"), ".json"), + ); const summary = { scanned: 0, reaped: 0, pruned: 0, active: 0, skipped: 0 }; if (stackFiles.length === 0) return { ...summary, durationMs: Date.now() - startedAt }; @@ -226,16 +235,15 @@ export const make = Effect.gen(function* DevStackReaperMake() { for (const file of stackFiles) { summary.scanned++; const raw = yield* Effect.promise(() => - fs - .readFile(file, "utf8") + NodeFSP.readFile(file, "utf8") .then(JSON.parse) .catch(() => null), ); const decoded = raw === null ? Option.none() : decodeEntry(raw); if (Option.isNone(decoded)) { // Not a shape we understand. Only bookkeeping is removed; no signals sent. - yield* Effect.promise(() => fs.rm(file, { force: true })); - yield* Effect.promise(() => fs.rm(seenPathFor(file), { force: true })); + yield* Effect.promise(() => NodeFSP.rm(file, { force: true })); + yield* Effect.promise(() => NodeFSP.rm(seenPathFor(file), { force: true })); summary.pruned++; continue; } @@ -262,14 +270,12 @@ export const make = Effect.gen(function* DevStackReaperMake() { } const anyProcessAlive = entry.processes.some((process_) => alive(process_.pid)); const rootExists = yield* Effect.promise(() => - fs - .stat(entry.root) + NodeFSP.stat(entry.root) .then(() => true) .catch(() => false), ); const lastSeenMs = yield* Effect.promise(() => - fs - .stat(seenPathFor(file)) + NodeFSP.stat(seenPathFor(file)) .then((stat) => stat.mtimeMs) .catch(() => null), ); @@ -288,8 +294,8 @@ export const make = Effect.gen(function* DevStackReaperMake() { case "StartClock": { const when = new Date(); yield* Effect.promise(async () => { - await fs.writeFile(seenPathFor(file), "").catch(() => {}); - await fs.utimes(seenPathFor(file), when, when).catch(() => {}); + await NodeFSP.writeFile(seenPathFor(file), "").catch(() => {}); + await NodeFSP.utimes(seenPathFor(file), when, when).catch(() => {}); }); summary.active++; break; diff --git a/packages/shared/src/t3ProjectFile.test.ts b/packages/shared/src/t3ProjectFile.test.ts index 7b9c2a31d9d..09297d85ec5 100644 --- a/packages/shared/src/t3ProjectFile.test.ts +++ b/packages/shared/src/t3ProjectFile.test.ts @@ -22,14 +22,26 @@ describe("buildT3ProjectFileJsonSchema", () => { { description?: string; items?: { properties: Record; required: ReadonlyArray }; + properties?: Record; } >; required?: ReadonlyArray; }; - expect(Object.keys(schema.properties).sort()).toEqual(["$schema", "iconPath", "scripts"]); + expect(Object.keys(schema.properties).sort()).toEqual([ + "$schema", + "devStacks", + "iconPath", + "scripts", + ]); expect(schema.required).toBeUndefined(); expect(schema.properties.iconPath?.description).toContain("Workspace-relative path"); + expect(schema.properties.devStacks?.description).toContain("supervising the dev stacks"); + expect(Object.keys(schema.properties.devStacks?.properties ?? {}).sort()).toEqual([ + "consumers", + "entryRoles", + "idleMinutes", + ]); const script = schema.properties.scripts?.items; expect(script?.required).toEqual(["name", "command"]);