diff --git a/services/cloud-agent-next/src/persistence/SandboxControl.ts b/services/cloud-agent-next/src/persistence/SandboxControl.ts index a08bdb2397..177715a7ae 100644 --- a/services/cloud-agent-next/src/persistence/SandboxControl.ts +++ b/services/cloud-agent-next/src/persistence/SandboxControl.ts @@ -3340,11 +3340,14 @@ export class SandboxControl extends DurableObject { 1_000, 'Diagnostic signing secret lookup timed out' ).catch(() => null); + const workloadCgroup = (this.env as { CONTROL_WORKLOAD_CGROUP?: unknown }) + .CONTROL_WORKLOAD_CGROUP; const launchEnv = buildControlWrapperLaunchEnv({ workerUrl: this.env.WORKER_URL, sandboxId: this.sandboxId, credential, diagnostics: { allocationId, signingSecret }, + ...(typeof workloadCgroup === 'string' ? { workloadCgroup } : {}), }); this.logDiagnostic('wrapper_log_upload', { allocationId, diff --git a/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.test.ts b/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.test.ts index f86f432c26..27741f4283 100644 --- a/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.test.ts +++ b/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.test.ts @@ -90,6 +90,22 @@ describe('buildControlWrapperLaunchEnv', () => { }); }); + it('forwards a non-empty workload cgroup and omits an absent or empty one', () => { + const base = { + workerUrl: 'https://worker.example.com', + sandboxId: 'sbx_1', + credential: 'control-credential', + }; + expect(buildControlWrapperLaunchEnv(base)).not.toHaveProperty('CONTROL_WORKLOAD_CGROUP'); + expect( + buildControlWrapperLaunchEnv({ ...base, workloadCgroup: 'kilo-workloads' }) + .CONTROL_WORKLOAD_CGROUP + ).toBe('kilo-workloads'); + expect( + buildControlWrapperLaunchEnv({ ...base, workloadCgroup: '' }).CONTROL_WORKLOAD_CGROUP + ).toBeUndefined(); + }); + it.each([ { workerUrl: undefined, expectedBase: '' }, { diff --git a/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.ts b/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.ts index 2fb343b341..ae7673c693 100644 --- a/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.ts +++ b/services/cloud-agent-next/src/sandbox-control/wrapper-launch-env.ts @@ -5,6 +5,7 @@ export type ControlWrapperLaunchEnvInput = { workerUrl?: string; sandboxId: string; credential: string; + workloadCgroup?: string; diagnostics?: { allocationId: string; signingSecret: string | null; @@ -48,6 +49,7 @@ export function buildControlWrapperLaunchEnv( } return { ...diagnosticEnv, + ...(input.workloadCgroup ? { CONTROL_WORKLOAD_CGROUP: input.workloadCgroup } : {}), SANDBOX_CONTROL_URL: sandboxControlWebSocketUrl(workerUrl, input.sandboxId), SANDBOX_CONTROL_CREDENTIAL: input.credential, PROVIDER_INSTANCE_ID: input.sandboxId, diff --git a/services/cloud-agent-next/src/shared/control-diagnostics.ts b/services/cloud-agent-next/src/shared/control-diagnostics.ts index ed158a1b60..07cc935903 100644 --- a/services/cloud-agent-next/src/shared/control-diagnostics.ts +++ b/services/cloud-agent-next/src/shared/control-diagnostics.ts @@ -275,6 +275,41 @@ export const controlDiagnosticFieldsSchema = z statusQueryPending: z.boolean().optional(), questionQueryPending: z.boolean().optional(), permissionQueryPending: z.boolean().optional(), + workloadPhase: z + .enum(['probe', 'applied', 'migration', 'oom', 'stats', 'rollback', 'failed']) + .optional(), + workloadFailure: z + .enum([ + 'flag_off', + 'no_finite_limit', + 'below_minimum', + 'not_delegated', + 'occupied', + 'controller_unavailable', + 'readback_mismatch', + 'write_failed', + 'unavailable', + 'pid_changed', + 'membership_unconfirmed', + ]) + .optional(), + containerLimitBytes: count.optional(), + aggregateMaxBytes: count.optional(), + reserveBytes: count.optional(), + appliedMaxBytes: count.optional(), + readbackMaxBytes: count.optional(), + currentBytes: count.optional(), + peakBytes: count.optional(), + oomKills: count.optional(), + oomGroupKills: count.optional(), + migratedCount: count.optional(), + toolCount: count.optional(), + serverCount: count.optional(), + pressureSomeTotal: count.optional(), + pressureFullTotal: count.optional(), + cpuController: z.boolean().optional(), + siblingProtection: z.boolean().optional(), + workloadLimitSource: z.enum(['cgroup', 'explicit', 'meminfo']).optional(), }) .strict(); @@ -291,6 +326,7 @@ export const controlDiagnosticRecordSchema = z 'control.request', 'control.event', 'control.upload', + 'control.workload', ]), fields: controlDiagnosticFieldsSchema, }) diff --git a/services/cloud-agent-next/wrapper/src/control/main.ts b/services/cloud-agent-next/wrapper/src/control/main.ts index 274255fd46..ab7f2ae72a 100644 --- a/services/cloud-agent-next/wrapper/src/control/main.ts +++ b/services/cloud-agent-next/wrapper/src/control/main.ts @@ -36,6 +36,7 @@ import { } from '../../../src/shared/control-diagnostics.js'; import { createWorktreeMutationNotifications } from './worktree-mutation-notifications'; import { createControlEventFailureHandler } from './control-event-transport'; +import { closeControlWorkload, initializeControlWorkload } from './workload-cgroup'; import type { ControlEventOutboxFailure } from './control-event-outbox'; @@ -53,6 +54,10 @@ function main( delete process.env.SANDBOX_CONTROL_CREDENTIAL; logToFile(`control-plane wrapper ${WRAPPER_VERSION} starting`); + const workload = initializeControlWorkload({ + env: process.env, + report: diagnostics.onDiagnostic, + }); const abort = new AbortController(); let control: ReturnType = null; let shuttingDown = false; @@ -116,6 +121,7 @@ function main( deps.operations.notifyRootDisappeared(disappearance); }; const kiloRuntimes = createWorktreeKiloRuntimes({ + workload, onDiagnostic: diagnostics.onDiagnostic, onRootRetirementStarted: markRootRetirementStarted, onRootDisappeared: notifyRootDisappeared, @@ -379,6 +385,7 @@ function main( try { kiloRuntimes.shutdown(); } finally { + closeControlWorkload(workload); process.exit(exitCode); } } diff --git a/services/cloud-agent-next/wrapper/src/control/owned-processes.test.ts b/services/cloud-agent-next/wrapper/src/control/owned-processes.test.ts index 41fbb9cc40..31e9ff3db8 100644 --- a/services/cloud-agent-next/wrapper/src/control/owned-processes.test.ts +++ b/services/cloud-agent-next/wrapper/src/control/owned-processes.test.ts @@ -1,10 +1,26 @@ import { afterEach, describe, expect, it, spyOn } from 'bun:test'; +import type { ChildProcess } from 'node:child_process'; import { once } from 'node:events'; import * as fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; import { runProcess } from '../utils.js'; -import { createOwnedProcessScope, classifyDirectProcessState } from './owned-processes.js'; +import { + createOwnedProcessScope, + classifyDirectProcessState, + migrateWorkloadProcess, +} from './owned-processes.js'; +import { + closeControlWorkload, + initializeControlWorkload, + type ControlWorkload, + type WorkloadPlacement, + type WorkloadProcessEntry, +} from './workload-cgroup.js'; const spawned: ReturnType[] = []; +const workloads: ControlWorkload[] = []; +const temporaryDirectories: string[] = []; function killPid(pid: number): void { try { @@ -16,8 +32,82 @@ function killPid(pid: number): void { afterEach(async () => { await Promise.all(spawned.splice(0).map(scope => scope.stop(Date.now() + 1_000))); + for (const workload of workloads.splice(0)) closeControlWorkload(workload); + for (const directory of temporaryDirectories.splice(0)) { + try { + fs.rmSync(directory, { recursive: true, force: true }); + } catch { + // Best-effort test cleanup. + } + } }); +function workloadPlacement(): WorkloadPlacement | undefined { + if (process.platform !== 'linux') return undefined; + const workload = initializeControlWorkload({ env: { CONTROL_WORKLOAD_CGROUP: '1' } }); + workloads.push(workload); + return workload.placement; +} + +function findManagedScope(placement: WorkloadPlacement, before: Set): string | undefined { + try { + const name = fs + .readdirSync(placement.parentReference) + .find(candidate => candidate.startsWith('kilo-control-') && !before.has(candidate)); + return name ? path.join(placement.parentReference, name) : undefined; + } catch { + return undefined; + } +} + +function cgroupPids(pathname: string): number[] { + try { + return fs + .readFileSync(pathname, 'utf8') + .split('\n') + .filter(Boolean) + .map(line => Number(line)); + } catch { + return []; + } +} + +function recursiveCgroupPids(directory: string): number[] { + const pids: number[] = []; + const visit = (current: string): void => { + let entries; + try { + entries = fs.readdirSync(current, { withFileTypes: true }); + } catch { + return; + } + pids.push(...cgroupPids(path.join(current, 'cgroup.procs'))); + for (const entry of entries) { + if (entry.isDirectory()) visit(path.join(current, entry.name)); + } + }; + visit(directory); + return pids; +} + +async function waitFor(condition: () => boolean, timeoutMs = 10_000): Promise { + const deadlineAt = Date.now() + timeoutMs; + while (Date.now() < deadlineAt) { + if (condition()) return true; + await Bun.sleep(25); + } + return condition(); +} + +function processDead(pid: number): boolean { + try { + process.kill(pid, 0); + return false; + } catch { + return true; + } +} + const descendantThatExitsMs = (ms: number) => `const { spawn } = require('node:child_process'); const child = spawn(process.execPath, ['-e', 'setTimeout(() => {}, ${ms})'], { stdio: 'ignore' }); process.stdout.write(String(child.pid)); setTimeout(() => process.exit(0), 20);`; @@ -42,6 +132,168 @@ describe('owned process scopes', () => { if (process.platform !== 'linux') expect(await scope.verify(false)).toBe(false); }); + it('returns an unmanaged replacement when gate admission fails', async () => { + if (process.platform !== 'linux') return; + const membership = readSelfCgroupMembership(); + if (!membership) return; + const directory = path.join('/sys/fs/cgroup', membership); + let before: Set; + try { + before = new Set(fs.readdirSync(directory)); + } catch { + return; + } + const originalReadFileSync = fs.readFileSync; + const statFailure = spyOn(fs, 'readFileSync').mockImplementation((( + target: unknown, + ...rest: unknown[] + ): string | Buffer => { + if (typeof target === 'string' && /^\/proc\/\d+\/stat$/.test(target)) { + throw Object.assign(new Error('simulated stat read failure'), { code: 'EACCES' }); + } + return (originalReadFileSync as (value: unknown, ...args: unknown[]) => string | Buffer)( + target, + ...rest + ); + }) as typeof fs.readFileSync); + const scope = createOwnedProcessScope(); + spawned.push(scope); + const children: ChildProcess[] = []; + let created: string | undefined; + try { + const child = scope.spawn(process.execPath, ['-e', 'setInterval(() => {}, 1_000)'], { + cwd: process.cwd(), + env: process.env, + }); + children.push(child); + created = fs + .readdirSync(directory) + .find(name => name.startsWith('kilo-control-') && !before.has(name)); + if (!created) return; + expect(child.spawnfile).toBe(process.execPath); + expect(child.spawnfile).not.toBe('/bin/sh'); + expect(child.pid).toBeDefined(); + expect(scope.observesOccupancy()).toBe(false); + const later = scope.spawn(process.execPath, ['-e', 'setInterval(() => {}, 1_000)'], { + cwd: process.cwd(), + env: process.env, + }); + children.push(later); + expect(later.spawnfile).toBe(process.execPath); + expect(later.spawnfile).not.toBe('/bin/sh'); + await scope.stop(Date.now() + 5_000); + scope.releaseAbandoned(); + } finally { + statFailure.mockRestore(); + for (const child of children) { + if (child.pid !== undefined) killPid(child.pid); + } + if (created) { + try { + fs.rmdirSync(path.join(directory, created)); + } catch { + console.warn('Owned process gate failure cleanup failed'); + } + } + } + }); + + it('does not return the gated shell when a gate release write fails without placement', async () => { + if (process.platform !== 'linux') return; + const membership = readSelfCgroupMembership(); + if (!membership) return; + const directory = path.join('/sys/fs/cgroup', membership); + let before: Set; + try { + before = new Set(fs.readdirSync(directory)); + } catch { + return; + } + const originalWriteSync = fs.writeSync; + const gateFailure = spyOn(fs, 'writeSync').mockImplementation((( + target: unknown, + ...rest: unknown[] + ): number => { + if (rest[0] === 'start\n') { + throw Object.assign(new Error('simulated gate release failure'), { code: 'EPIPE' }); + } + return (originalWriteSync as (value: unknown, ...args: unknown[]) => number)(target, ...rest); + }) as typeof fs.writeSync); + const scope = createOwnedProcessScope(); + spawned.push(scope); + let created: string | undefined; + let replacement: ChildProcess | undefined; + try { + const child = scope.spawn(process.execPath, ['-e', 'setInterval(() => {}, 1_000)'], { + cwd: process.cwd(), + env: process.env, + }); + replacement = child; + created = fs + .readdirSync(directory) + .find(name => name.startsWith('kilo-control-') && !before.has(name)); + if (!created) return; + expect(child.spawnfile).toBe(process.execPath); + expect(child.spawnfile).not.toBe('/bin/sh'); + expect(child.pid).toBeDefined(); + expect(scope.observesOccupancy()).toBe(false); + await scope.stop(Date.now() + 5_000); + scope.releaseAbandoned(); + expect(await waitFor(() => processDead(child.pid as number))).toBe(true); + } finally { + gateFailure.mockRestore(); + if (replacement?.pid !== undefined) killPid(replacement.pid); + if (created) { + try { + fs.rmdirSync(path.join(directory, created)); + } catch { + console.warn('Owned process gate failure cleanup failed'); + } + } + } + }); + + it('returns an unmanaged replacement when the gate release write fails', async () => { + const placement = workloadPlacement(); + if (!placement) return; + const before = new Set( + fs.readdirSync(placement.parentReference).filter(name => name.startsWith('kilo-control-')) + ); + const originalWriteSync = fs.writeSync; + const gateFailure = spyOn(fs, 'writeSync').mockImplementation((( + target: unknown, + ...rest: unknown[] + ): number => { + if (rest[0] === 'start\n') { + throw Object.assign(new Error('simulated gate release failure'), { code: 'EPIPE' }); + } + return (originalWriteSync as (value: unknown, ...args: unknown[]) => number)(target, ...rest); + }) as typeof fs.writeSync); + const scope = createOwnedProcessScope(placement); + spawned.push(scope); + try { + const child = scope.spawn(process.execPath, ['-e', 'setInterval(() => {}, 1_000)'], { + cwd: process.cwd(), + env: process.env, + }); + expect(child.spawnfile).toBe(process.execPath); + expect(scope.observesOccupancy()).toBe(false); + await scope.stop(Date.now() + 5_000); + scope.releaseAbandoned(); + expect(await waitFor(() => processDead(child.pid as number))).toBe(true); + } finally { + gateFailure.mockRestore(); + const owned = findManagedScope(placement, before); + if (owned) { + try { + fs.rmSync(owned, { recursive: true, force: true }); + } catch { + console.warn('Owned process gate failure cleanup failed'); + } + } + } + }); + it('keeps proven death after a bounded cgroup removal failure on Linux', async () => { if (process.platform !== 'linux') return; const scope = createOwnedProcessScope(); @@ -307,3 +559,231 @@ describe('owned process scopes', () => { }); }); }); + +describe('workload migration step', () => { + const entry: WorkloadProcessEntry = { pid: 7, ppid: 1, argv: ['bash', '-c', 'build'] }; + + it('checks identity immediately before the write', async () => { + const calls: string[] = []; + const outcome = await migrateWorkloadProcess({ + pid: entry.pid, + entry, + procRoot: '/proc', + identityMatches: async () => { + calls.push('check'); + return true; + }, + write: () => { + calls.push('write'); + }, + confirmMembership: async () => true, + }); + expect(outcome).toBe('migrated'); + expect(calls).toEqual(['check', 'write']); + }); + + it('skips the write and reports pid_changed when the identity changed after selection', async () => { + const writes: number[] = []; + const outcome = await migrateWorkloadProcess({ + pid: entry.pid, + entry, + procRoot: '/proc', + identityMatches: async () => false, + write: pid => writes.push(pid), + confirmMembership: async () => true, + }); + expect(outcome).toBe('pid_changed'); + expect(writes).toEqual([]); + }); + + it('reports pid_changed without writing when the process left the snapshot', async () => { + const writes: number[] = []; + const outcome = await migrateWorkloadProcess({ + pid: entry.pid, + entry: undefined, + procRoot: '/proc', + write: pid => writes.push(pid), + confirmMembership: async () => true, + }); + expect(outcome).toBe('pid_changed'); + expect(writes).toEqual([]); + }); + + it('reports membership_unconfirmed when the write is not reflected', async () => { + const outcome = await migrateWorkloadProcess({ + pid: entry.pid, + entry, + procRoot: '/proc', + identityMatches: async () => true, + write: () => undefined, + confirmMembership: async () => false, + }); + expect(outcome).toBe('membership_unconfirmed'); + }); +}); + +function readSelfCgroupMembership(): string | undefined { + try { + const line = fs + .readFileSync('/proc/self/cgroup', 'utf8') + .split('\n') + .find(candidate => candidate.startsWith('0::')); + const membership = line?.slice(3); + return membership && membership.startsWith('/') ? membership : undefined; + } catch { + return undefined; + } +} + +describe('workload cgroup integration', () => { + it('keeps a managed scope under the workload parent and splits server from tools', async () => { + const placement = workloadPlacement(); + if (!placement) return; + const before = new Set( + fs.readdirSync(placement.parentReference).filter(name => name.startsWith('kilo-control-')) + ); + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'workload-owned-')); + temporaryDirectories.push(directory); + const toolPidFile = path.join(directory, 'tool.pid'); + const kilo = path.join(directory, 'kilo'); + fs.writeFileSync(kilo, `#!/bin/sh\nsleep 30 &\necho $! > ${toolPidFile}\nwait\n`, { + mode: 0o755, + }); + + const scope = createOwnedProcessScope(placement); + spawned.push(scope); + const server = scope.spawn(kilo, ['serve', '--port=0'], { cwd: directory, env: process.env }); + if (!scope.observesOccupancy()) return; + const serverPid = server.pid; + expect(serverPid).toBeDefined(); + if (serverPid === undefined) return; + + const owned = findManagedScope(placement, before); + expect(owned).toBeDefined(); + if (!owned) return; + + expect( + await waitFor(() => + cgroupPids(path.join(owned, 'server', 'cgroup.procs')).includes(serverPid) + ) + ).toBe(true); + expect( + await waitFor(() => { + try { + return Number(fs.readFileSync(toolPidFile, 'utf8').trim()) > 0; + } catch { + return false; + } + }) + ).toBe(true); + const toolPid = Number(fs.readFileSync(toolPidFile, 'utf8').trim()); + expect( + await waitFor(() => cgroupPids(path.join(owned, 'tools', 'cgroup.procs')).includes(toolPid)) + ).toBe(true); + + expect(cgroupPids(path.join(owned, 'server', 'cgroup.procs'))).not.toContain(toolPid); + expect(recursiveCgroupPids(owned)).toEqual(expect.arrayContaining([serverPid, toolPid])); + expect( + fs.readFileSync(path.join(placement.parentReference, 'memory.oom.group'), 'utf8').trim() + ).toBe('0'); + expect(await scope.verify(false)).toBe(false); + + expect(await scope.stop(Date.now() + 5_000)).toBe(true); + expect(await waitFor(() => processDead(serverPid) && processDead(toolPid))).toBe(true); + expect(await waitFor(() => !fs.existsSync(owned))).toBe(true); + }); + + it('moves a kilo serve process born in the tools group back to server', async () => { + const placement = workloadPlacement(); + if (!placement) return; + const before = new Set( + fs.readdirSync(placement.parentReference).filter(name => name.startsWith('kilo-control-')) + ); + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'workload-owned-')); + temporaryDirectories.push(directory); + const kilo = path.join(directory, 'kilo'); + const tool = path.join(directory, 'tool.sh'); + const nestedPidFile = path.join(directory, 'nested.pid'); + fs.writeFileSync( + tool, + `#!/bin/sh\nwhile ! grep -q /tools /proc/self/cgroup; do sleep 0.1; done\n"${kilo}" serve --from-tool &\necho $! > "${nestedPidFile}"\nsleep 30\n`, + { mode: 0o755 } + ); + fs.writeFileSync( + kilo, + `#!/bin/sh\nif [ "$2" = --from-tool ]; then sleep 30; exit; fi\n"${tool}" &\nwait\n`, + { mode: 0o755 } + ); + const scope = createOwnedProcessScope(placement); + spawned.push(scope); + const server = scope.spawn(kilo, ['serve', '--port=0'], { cwd: directory, env: process.env }); + if (!scope.observesOccupancy()) return; + const owned = findManagedScope(placement, before); + expect(owned).toBeDefined(); + if (!owned) return; + expect( + await waitFor(() => { + try { + return Number(fs.readFileSync(nestedPidFile, 'utf8').trim()) > 0; + } catch { + return false; + } + }) + ).toBe(true); + const nestedPid = Number(fs.readFileSync(nestedPidFile, 'utf8').trim()); + expect( + await waitFor(() => + cgroupPids(path.join(owned, 'server', 'cgroup.procs')).includes(nestedPid) + ) + ).toBe(true); + expect(cgroupPids(path.join(owned, 'tools', 'cgroup.procs'))).not.toContain(nestedPid); + expect(server.pid).toBeDefined(); + }); + + it('does not create workload children for an unmanaged scope', async () => { + if (process.platform !== 'linux') return; + const membership = readSelfCgroupMembership(); + if (!membership) return; + const directory = path.join('/sys/fs/cgroup', membership); + const before = new Set( + fs.readdirSync(directory).filter(name => name.startsWith('kilo-control-')) + ); + const scope = createOwnedProcessScope(); + spawned.push(scope); + const child = scope.spawn(process.execPath, ['-e', 'process.exit(0)'], { + cwd: process.cwd(), + env: process.env, + }); + await once(child, 'exit'); + if (!scope.observesOccupancy()) return; + for (const name of fs.readdirSync(directory)) { + if (!name.startsWith('kilo-control-') || before.has(name)) continue; + expect(fs.existsSync(path.join(directory, name, 'server'))).toBe(false); + expect(fs.existsSync(path.join(directory, name, 'tools'))).toBe(false); + } + }); + + it('reports the tools cpu read-back in the managed scope stats diagnostic', async () => { + if (process.platform !== 'linux') return; + const reports: Array<{ event: string; fields: Record }> = []; + const workload = initializeControlWorkload({ + env: { CONTROL_WORKLOAD_CGROUP: '1' }, + report: (event, fields) => reports.push({ event, fields }), + }); + workloads.push(workload); + const placement = workload.placement; + if (!placement) return; + const scope = createOwnedProcessScope(placement); + spawned.push(scope); + scope.spawn(process.execPath, ['-e', 'setInterval(() => {}, 1_000)'], { + cwd: process.cwd(), + env: process.env, + }); + if (!scope.observesOccupancy()) return; + await waitFor(() => reports.some(entry => entry.fields.workloadPhase === 'stats')); + const stats = reports.find(entry => entry.fields.workloadPhase === 'stats'); + expect(stats?.event).toBe('control.workload'); + expect(typeof stats?.fields.cpuController).toBe('boolean'); + expect(await scope.stop(Date.now() + 1_000)).toBe(true); + }); +}); diff --git a/services/cloud-agent-next/wrapper/src/control/owned-processes.ts b/services/cloud-agent-next/wrapper/src/control/owned-processes.ts index 922f25a19f..dd2fb7dd4a 100644 --- a/services/cloud-agent-next/wrapper/src/control/owned-processes.ts +++ b/services/cloud-agent-next/wrapper/src/control/owned-processes.ts @@ -22,6 +22,19 @@ import { lstat, readFile, readdir } from 'node:fs/promises'; import path from 'node:path'; import { Writable } from 'node:stream'; import { setTimeout as delay } from 'node:timers/promises'; +import { pidStillMatches, readProcessTable } from '../tool-cgroup.js'; +import { + applyManagedWorkloadLimits, + CGROUP_FS_MAGIC, + classifyWorkloadMembers, + createWorkloadReporter, + readWorkloadStats, + WORKLOAD_SERVER_NAME, + WORKLOAD_SWEEP_INTERVAL_MS, + WORKLOAD_TOOLS_NAME, + type WorkloadPlacement, + type WorkloadProcessEntry, +} from './workload-cgroup.js'; export type DirectProcessState = 'absent' | 'reused' | 'alive' | 'unknown'; @@ -66,7 +79,15 @@ type Cgroup = { ino: number; descriptors: number[]; procs: number; + procsReference: string; kill?: number; + managed?: { + serverReference: string; + toolsReference: string; + toolsProcs: number; + serverProcs: number; + cpuController: boolean; + }; }; export const OWNED_PROCESS_OBSERVATION_TIMEOUT_MS = 1_000; @@ -232,6 +253,13 @@ function releaseChildStreams(child: OwnedChild): void { } } +function releaseGate(gate: Writable): void { + const fd = (gate as Writable & { _handle?: { fd?: number } })._handle?.fd; + if (typeof fd !== 'number') throw new Error('Owned child gate unavailable'); + writeSync(fd, 'start\n'); + gate.destroy(); +} + function isErofs(error: unknown): boolean { return typeof error === 'object' && error !== null && 'code' in error && error.code === 'EROFS'; } @@ -315,7 +343,7 @@ function createCgroup(): Cgroup | undefined { console.warn('Owned process cgroup.kill unavailable; using verified child signals'); } closeDescriptors(descriptors.splice(0, 2)); - return { directory, reference, dev, ino, descriptors, procs, kill }; + return { directory, reference, dev, ino, descriptors, procs, procsReference: reference, kill }; } catch (error) { const message = error instanceof Error ? error.message : 'unknown'; console.warn(`Owned process containment unavailable: ${message}`); @@ -336,6 +364,155 @@ function sameDirectory(group: Cgroup, value: { dev: number; ino: number }): bool return value.dev === group.dev && value.ino === group.ino; } +export type WorkloadMigrationOutcome = + | 'migrated' + | 'pid_changed' + | 'write_failed' + | 'membership_unconfirmed'; + +export async function migrateWorkloadProcess(input: { + pid: number; + entry: WorkloadProcessEntry | undefined; + procRoot: string; + write: (pid: number) => void; + confirmMembership: () => Promise; + identityMatches?: (procRoot: string, entry: WorkloadProcessEntry) => Promise; +}): Promise { + const matches = input.identityMatches ?? pidStillMatches; + if (!input.entry || !(await matches(input.procRoot, input.entry))) return 'pid_changed'; + try { + input.write(input.pid); + } catch { + return 'write_failed'; + } + try { + return (await input.confirmMembership()) ? 'migrated' : 'membership_unconfirmed'; + } catch { + return 'membership_unconfirmed'; + } +} + +function removeDirectories(directory: string): void { + let entries; + try { + entries = readdirSync(directory, { withFileTypes: true }); + } catch { + return; + } + for (const entry of entries) { + if (!entry.isDirectory() || entry.isSymbolicLink()) continue; + removeDirectories(path.join(directory, entry.name)); + } + try { + rmdirSync(directory); + } catch { + console.warn('Owned process containment creation cleanup failed'); + } +} + +function createManagedCgroup(placement: WorkloadPlacement): Cgroup | undefined { + const descriptors: number[] = []; + let created: { directory: string; dev: number; ino: number } | undefined; + try { + if (process.platform !== 'linux') return undefined; + if (statfsSync(placement.parentReference).type !== CGROUP_FS_MAGIC) { + throw new Error('Workload parent is not a cgroup'); + } + const parent = fstatSync(placement.parentFd); + if (parent.dev !== placement.parentDev || parent.ino !== placement.parentIno) { + throw new Error('Workload parent changed'); + } + const name = `kilo-control-${crypto.randomUUID()}`; + try { + mkdirSync(path.join(placement.parentReference, name)); + } catch (error) { + if (!isErofs(error)) throw error; + remountCgroupWritable(placement.parentDirectory); + mkdirSync(path.join(placement.parentReference, name)); + } + const directory = path.join(placement.parentDirectory, name); + const descriptor = openSync( + path.join(placement.parentReference, name), + constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW + ); + descriptors.push(descriptor); + const { dev, ino } = fstatSync(descriptor); + created = { directory, dev, ino }; + const reference = `/proc/self/fd/${descriptor}`; + if (population(readFileSync(path.join(reference, 'cgroup.events'), 'utf8')) !== 0) { + throw new Error('Owned process containment is occupied'); + } + + mkdirSync(path.join(reference, WORKLOAD_SERVER_NAME)); + mkdirSync(path.join(reference, WORKLOAD_TOOLS_NAME)); + const serverFd = openSync( + path.join(reference, WORKLOAD_SERVER_NAME), + constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW + ); + descriptors.push(serverFd); + const toolsFd = openSync( + path.join(reference, WORKLOAD_TOOLS_NAME), + constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW + ); + descriptors.push(toolsFd); + const serverReference = `/proc/self/fd/${serverFd}`; + const toolsReference = `/proc/self/fd/${toolsFd}`; + + const { cpuController } = applyManagedWorkloadLimits({ + parentReference: reference, + serverReference, + toolsReference, + aggregateMaxBytes: placement.aggregateMaxBytes, + }); + + const procs = openSync( + path.join(serverReference, 'cgroup.procs'), + constants.O_WRONLY | constants.O_NOFOLLOW + ); + descriptors.push(procs); + const toolsProcs = openSync( + path.join(toolsReference, 'cgroup.procs'), + constants.O_WRONLY | constants.O_NOFOLLOW + ); + descriptors.push(toolsProcs); + let kill: number | undefined; + try { + kill = openSync( + path.join(reference, 'cgroup.kill'), + constants.O_WRONLY | constants.O_NOFOLLOW + ); + descriptors.push(kill); + } catch { + console.warn('Owned process cgroup.kill unavailable; using verified child signals'); + } + return { + directory, + reference, + dev, + ino, + descriptors, + procs, + procsReference: serverReference, + managed: { serverReference, toolsReference, toolsProcs, serverProcs: procs, cpuController }, + ...(kill !== undefined ? { kill } : {}), + }; + } catch (error) { + const message = error instanceof Error ? error.message : 'unknown'; + console.warn(`Owned process workload containment unavailable: ${message}`); + if (created) { + try { + const fresh = lstatSync(created.directory); + if (fresh.dev === created.dev && fresh.ino === created.ino) + removeDirectories(created.directory); + } catch { + console.warn('Owned process containment creation cleanup failed'); + } + } + closeDescriptors(descriptors); + return undefined; + } +} + async function assertDirectory(group: Cgroup, deadline: Deadline): Promise { if (group.descriptors.length === 0) throw new Error('Owned process containment is closed'); const value = await deadline.wait(() => lstat(group.directory)); @@ -437,8 +614,9 @@ function removeCgroup(group: Cgroup, deadlineAt: number): boolean { } } -export function createOwnedProcessScope(): OwnedProcessScope { +export function createOwnedProcessScope(placement?: WorkloadPlacement): OwnedProcessScope { let group: Cgroup | undefined; + let activePlacement = placement; let attempted = false; let contained = true; let sealed = false; @@ -450,14 +628,142 @@ export function createOwnedProcessScope(): OwnedProcessScope { let stopped = false; let abandoned = false; let cgroupRemoval: Promise | undefined; + let sweepTimer: ReturnType | undefined; + let sweeping = false; + let lastOomKills = 0; + let lastOomGroupKills = 0; const children = new Set(); const baseline = new Set(); const observations = new Set(); + const workloadReporter = placement?.report ? createWorkloadReporter(placement.report) : undefined; const live = (child: OwnedChild): boolean => !child.exited && child.process.pid !== undefined; const occupancyGroup = (): Cgroup | undefined => group !== undefined && contained ? group : undefined; const occupancyObservable = (): boolean => occupancyGroup() !== undefined; + const stopSweep = (): void => { + if (sweepTimer !== undefined) { + clearInterval(sweepTimer); + sweepTimer = undefined; + } + }; + + const sweepManaged = async (): Promise => { + const managed = group?.managed; + if (sweeping || !managed || !group || removed || stopped) return; + sweeping = true; + const deadline = createDeadline(Date.now() + OBSERVATION_TIMEOUT_MS); + observations.add(deadline); + try { + const root = [...children].find(live)?.process.pid; + if (root === undefined) return; + const scopeId = path.basename(group.directory); + const table = await readProcessTable('/proc'); + const snapshot = await snapshotCgroup(group, deadline); + const { serverPids, toolPids } = classifyWorkloadMembers(snapshot.pids, table, root); + const migrateInto = async ( + pids: number[], + members: Set, + procs: number, + reference: string + ): Promise => { + let moved = 0; + for (const pid of pids) { + if (members.has(pid)) continue; + deadline.check(); + const outcome = await migrateWorkloadProcess({ + pid, + entry: table.get(pid), + procRoot: '/proc', + write: value => writeSync(procs, String(value), 0, 'utf8'), + confirmMembership: async () => { + const after = new Set( + parsePids(await readText(path.join(reference, 'cgroup.procs'), deadline)) + ); + return after.has(pid); + }, + }); + if (outcome === 'migrated') { + moved += 1; + continue; + } + workloadReporter?.emit(scopeId, { + phase: 'failed', + workloadPhase: 'migration', + workloadFailure: outcome, + }); + } + return moved; + }; + const serverMembers = new Set( + parsePids(await readText(path.join(managed.serverReference, 'cgroup.procs'), deadline)) + ); + const toolsMembers = new Set( + parsePids(await readText(path.join(managed.toolsReference, 'cgroup.procs'), deadline)) + ); + const migrated = + (await migrateInto( + serverPids, + serverMembers, + managed.serverProcs, + managed.serverReference + )) + + (await migrateInto(toolPids, toolsMembers, managed.toolsProcs, managed.toolsReference)); + if (migrated > 0) { + workloadReporter?.emit(scopeId, { + phase: 'completed', + workloadPhase: 'migration', + migratedCount: migrated, + }); + } + const stats = readWorkloadStats(group.reference); + if (stats.oomKills > lastOomKills || stats.oomGroupKills > lastOomGroupKills) { + lastOomKills = Math.max(lastOomKills, stats.oomKills); + lastOomGroupKills = Math.max(lastOomGroupKills, stats.oomGroupKills); + workloadReporter?.emit(scopeId, { + phase: 'failed', + workloadPhase: 'oom', + oomKills: stats.oomKills, + oomGroupKills: stats.oomGroupKills, + }); + } + workloadReporter?.emit(scopeId, { + phase: 'completed', + workloadPhase: 'stats', + toolCount: toolPids.length, + serverCount: serverPids.length, + migratedCount: migrated, + oomKills: stats.oomKills, + oomGroupKills: stats.oomGroupKills, + cpuController: managed.cpuController, + ...(stats.currentBytes !== undefined ? { currentBytes: stats.currentBytes } : {}), + ...(stats.peakBytes !== undefined ? { peakBytes: stats.peakBytes } : {}), + ...(stats.pressureSomeTotal !== undefined + ? { pressureSomeTotal: stats.pressureSomeTotal } + : {}), + ...(stats.pressureFullTotal !== undefined + ? { pressureFullTotal: stats.pressureFullTotal } + : {}), + }); + } catch { + workloadReporter?.emit(path.basename(group.directory), { + phase: 'failed', + workloadPhase: 'migration', + workloadFailure: 'unavailable', + }); + } finally { + sweeping = false; + observations.delete(deadline); + deadline.close(); + } + }; + + const startSweep = (): void => { + if (sweepTimer !== undefined || !activePlacement) return; + sweepTimer = setInterval(() => void sweepManaged(), WORKLOAD_SWEEP_INTERVAL_MS); + sweepTimer.unref?.(); + }; + const verify = async (allowBaseline: boolean, deadline: Deadline): Promise => { try { deadline.check(); @@ -524,90 +830,119 @@ export function createOwnedProcessScope(): OwnedProcessScope { const scope: OwnedProcessScope = { spawn(command, args, options) { if (sealed) throw new Error('Owned process admission is closed'); + if (activePlacement && options.shell !== undefined && options.shell !== false) { + throw new Error('Owned process placement rejects shell spawn'); + } if (!attempted) { attempted = true; - group = createCgroup(); + group = activePlacement ? createManagedCgroup(activePlacement) : createCgroup(); + } + if (activePlacement && !group) { + workloadReporter?.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: 'unavailable', + }); + activePlacement = undefined; } const gated = group !== undefined && options.shell !== true && typeof options.shell !== 'string'; - const child = gated - ? spawn( - '/bin/sh', - [ - '-c', - 'IFS= read -r start <&3 && [ "$start" = start ] && exec 3<&- && exec "$@"', - 'kilo-owned', - command, - ...args, - ], - { ...options, detached: true, stdio: ['pipe', 'pipe', 'pipe', 'pipe'] } - ) - : spawn(command, args, { ...options, detached: true, stdio: 'pipe' }); - const record: OwnedChild = { process: child, exited: false }; - children.add(record); - child.once('exit', () => { - record.exited = true; - }); - child.once('error', () => { - if (child.pid === undefined) record.exited = true; - }); - if (child.pid !== undefined) { - used = true; - try { - const fresh = processIdentity(child.pid, readFileSync(`/proc/${child.pid}/stat`, 'utf8')); - if (fresh.parent === process.pid) record.identity = fresh.identity; - } catch { - console.warn('Owned process child identity unavailable; group signals remain disabled'); + + const spawnChild = (gatedChild: boolean): OwnedChild => { + const child = gatedChild + ? spawn( + '/bin/sh', + [ + '-c', + 'IFS= read -r start <&3 && [ "$start" = start ] && exec 3<&- && exec "$@"', + 'kilo-owned', + command, + ...args, + ], + { ...options, detached: true, stdio: ['pipe', 'pipe', 'pipe', 'pipe'] } + ) + : spawn(command, args, { ...options, detached: true, stdio: 'pipe' }); + const record: OwnedChild = { process: child, exited: false }; + children.add(record); + child.once('exit', () => { + record.exited = true; + }); + child.once('error', () => { + if (child.pid === undefined) record.exited = true; + }); + if (child.pid !== undefined) { + used = true; + try { + const fresh = processIdentity( + child.pid, + readFileSync(`/proc/${child.pid}/stat`, 'utf8') + ); + if (fresh.parent === process.pid) record.identity = fresh.identity; + } catch { + console.warn('Owned process child identity unavailable; group signals remain disabled'); + } + if (!gatedChild) contained = false; } - if (!gated) contained = false; + return record; + }; + + const record = spawnChild(gated); + if (!gated) { + startSweep(); + return record.process; } - if (gated) { - const gate = child.stdio[3]; - if (gate instanceof Writable) record.gate = gate; + const gate = record.process.stdio[3]; + if (gate instanceof Writable) record.gate = gate; + try { + const pid = record.process.pid; + if ( + !(gate instanceof Writable) || + !group || + pid === undefined || + !record.identity || + !sameDirectory(group, lstatSync(group.directory)) + ) { + throw new Error('Owned child identity unavailable'); + } + const fresh = processIdentity(pid, readFileSync(`/proc/${pid}/stat`, 'utf8')); + if (fresh.identity !== record.identity || fresh.parent !== process.pid) { + throw new Error('Owned child changed'); + } + writeSync(group.procs, String(pid), 0, 'utf8'); + if ( + !sameDirectory(group, lstatSync(group.directory)) || + !parsePids( + readFileSync(path.join(group.procsReference, 'cgroup.procs'), 'utf8') + ).includes(pid) + ) { + throw new Error('Owned child containment unavailable'); + } + const activated = processIdentity(pid, readFileSync(`/proc/${pid}/stat`, 'utf8')); + if (activated.identity !== record.identity || activated.parent !== process.pid) { + throw new Error('Owned child changed'); + } + releaseGate(gate); + } catch { + contained = false; + if (gate instanceof Writable) { + gate.on('error', () => undefined); + gate.end(); + } + activePlacement = undefined; + workloadReporter?.emit(group ? path.basename(group.directory) : undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: 'unavailable', + }); try { - if ( - !(gate instanceof Writable) || - !group || - child.pid === undefined || - !record.identity || - !sameDirectory(group, lstatSync(group.directory)) - ) { - throw new Error('Owned child identity unavailable'); - } - const fresh = processIdentity(child.pid, readFileSync(`/proc/${child.pid}/stat`, 'utf8')); - if (fresh.identity !== record.identity || fresh.parent !== process.pid) { - throw new Error('Owned child changed'); - } - writeSync(group.procs, String(child.pid), 0, 'utf8'); - if ( - !sameDirectory(group, lstatSync(group.directory)) || - !parsePids(readFileSync(path.join(group.reference, 'cgroup.procs'), 'utf8')).includes( - child.pid - ) - ) { - throw new Error('Owned child containment unavailable'); - } - const activated = processIdentity( - child.pid, - readFileSync(`/proc/${child.pid}/stat`, 'utf8') - ); - if (activated.identity !== record.identity || activated.parent !== process.pid) { - throw new Error('Owned child changed'); - } - gate.on('error', () => { - contained = false; - }); - gate.end('start\n'); + record.process.kill('SIGKILL'); } catch { - contained = false; - sealed = true; - if (gate instanceof Writable) { - gate.on('error', () => undefined); - gate.end(); - } + console.warn('Owned process gated child kill failed; continuing unmanaged'); } + return spawnChild(false).process; } - return child; + startSweep(); + return record.process; }, observeChild(child) { const record = [...children].find(candidate => candidate.process === child); @@ -667,6 +1002,7 @@ export function createOwnedProcessScope(): OwnedProcessScope { releaseAbandoned() { if (stopped || removed || abandoned) return; abandoned = true; + stopSweep(); for (const child of children) releaseChildStreams(child); if (group) closeDescriptors(group.descriptors); }, @@ -677,6 +1013,7 @@ export function createOwnedProcessScope(): OwnedProcessScope { }, dispose() { sealed = true; + stopSweep(); if (removed) return true; if (stopped) return true; if (abandoned) return false; @@ -759,6 +1096,7 @@ export function createOwnedProcessScope(): OwnedProcessScope { }, stop(deadlineAt) { sealed = true; + stopSweep(); for (const observation of observations) observation.shorten(deadlineAt); if (stopping) { if (!stopped) stopDeadline?.shorten(deadlineAt); diff --git a/services/cloud-agent-next/wrapper/src/control/workload-cgroup.test.ts b/services/cloud-agent-next/wrapper/src/control/workload-cgroup.test.ts new file mode 100644 index 0000000000..5f1726cbfd --- /dev/null +++ b/services/cloud-agent-next/wrapper/src/control/workload-cgroup.test.ts @@ -0,0 +1,923 @@ +import { afterEach, describe, expect, it } from 'bun:test'; +import { + chmodSync, + closeSync, + constants, + existsSync, + fstatSync, + mkdirSync, + mkdtempSync, + openSync, + readFileSync, + rmSync, + symlinkSync, + writeFileSync, +} from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { + controlDiagnosticRecordSchema, + createControlDiagnosticRecord, +} from '../../../src/shared/control-diagnostics.js'; +import { createOwnedProcessScope } from './owned-processes.js'; +import { + admitControlWorkload, + applyManagedWorkloadLimits, + classifyWorkloadMembers, + closeControlWorkload, + createWorkloadReporter, + computeWorkloadBudget, + initializeControlWorkload, + parsePressureTotal, + type ControlWorkload, + type WorkloadPlacement, + type WorkloadProcessEntry, +} from './workload-cgroup.js'; + +const MIB = 1024 * 1024; +const GIB = 1024 * MIB; +const RESERVE_BYTES = 2048 * MIB; + +const roots: string[] = []; +const placements: ControlWorkload[] = []; + +function makeRoot(): string { + const root = mkdtempSync(path.join(os.tmpdir(), 'workload-cgroup-')); + roots.push(root); + return root; +} + +function writeControl(directory: string, name: string, value: string): void { + mkdirSync(directory, { recursive: true }); + writeFileSync(path.join(directory, name), value); +} + +function selfCgroupFile(root: string, membership: string): string { + const file = path.join(root, 'self-cgroup'); + writeFileSync(file, `0::${membership}\n`); + return file; +} + +function runInitialize(input: { + root: string; + membership: string; + env: Record; + procRoot?: string; + platform?: NodeJS.Platform; +}): ControlWorkload { + const workload = initializeControlWorkload({ + env: input.env, + cgroupRoot: input.root, + selfCgroupFile: selfCgroupFile(input.root, input.membership), + procRoot: input.procRoot ?? path.join(input.root, 'proc'), + isCgroupMount: () => true, + handlePath: (_descriptor, directory) => directory, + platform: input.platform ?? 'linux', + }); + placements.push(workload); + return workload; +} + +function procWithMemTotal(root: string, memTotalKb: number): string { + const procRoot = path.join(root, 'proc'); + mkdirSync(procRoot, { recursive: true }); + writeFileSync(path.join(procRoot, 'meminfo'), `MemTotal:\t${memTotalKb} kB\n`); + return procRoot; +} + +function initialize( + root: string, + membership: string, + env: Record = {}, + procRoot?: string +): ControlWorkload { + return runInitialize({ + root, + membership, + env: { CONTROL_WORKLOAD_CGROUP: '1', ...env }, + ...(procRoot === undefined ? {} : { procRoot }), + }); +} + +function delegationTree(root: string): void { + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory cpu'); + writeControl(path.join(root, 'a'), 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + writeControl(path.join(root, 'a', 'kilo-workloads'), 'cgroup.events', 'populated 0'); +} + +afterEach(() => { + for (const workload of placements.splice(0)) closeControlWorkload(workload); + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +describe('workload budget', () => { + it('reserves the fixed control budget for the 4, 6 and 12 GiB classes', () => { + const cases: [number, number][] = [ + [4 * GIB, 2048 * MIB], + [6 * GIB, 4096 * MIB], + [12 * GIB, 10240 * MIB], + ]; + for (const [limit, expected] of cases) { + expect(computeWorkloadBudget({ limits: [limit], reserveBytes: RESERVE_BYTES })).toEqual({ + ok: true, + containerLimitBytes: limit, + aggregateMaxBytes: expected, + source: 'cgroup', + }); + } + }); + + it('reports no_finite_limit when no governing ancestor has a finite limit', () => { + expect( + computeWorkloadBudget({ limits: [undefined, undefined], reserveBytes: RESERVE_BYTES }) + ).toEqual({ ok: false, failure: 'no_finite_limit' }); + expect(computeWorkloadBudget({ limits: [], reserveBytes: RESERVE_BYTES })).toEqual({ + ok: false, + failure: 'no_finite_limit', + }); + }); + + it('reports below_minimum when the aggregate would fall under 1 GiB', () => { + expect(computeWorkloadBudget({ limits: [2560 * MIB], reserveBytes: RESERVE_BYTES })).toEqual({ + ok: false, + failure: 'below_minimum', + }); + }); + + it('takes the tighter of the ancestor limit and the explicit limit', () => { + expect( + computeWorkloadBudget({ + limits: [8 * GIB], + explicitLimitBytes: 4 * GIB, + reserveBytes: RESERVE_BYTES, + }) + ).toEqual({ + ok: true, + containerLimitBytes: 4 * GIB, + aggregateMaxBytes: 2048 * MIB, + source: 'explicit', + }); + expect( + computeWorkloadBudget({ + limits: [4 * GIB], + explicitLimitBytes: 8 * GIB, + reserveBytes: RESERVE_BYTES, + }) + ).toEqual({ + ok: true, + containerLimitBytes: 4 * GIB, + aggregateMaxBytes: 2048 * MIB, + source: 'cgroup', + }); + }); + + it('prefers a finite cgroup limit over a larger meminfo total', () => { + expect( + computeWorkloadBudget({ + limits: [4 * GIB], + memTotalBytes: 8 * GIB, + reserveBytes: RESERVE_BYTES, + }) + ).toEqual({ + ok: true, + containerLimitBytes: 4 * GIB, + aggregateMaxBytes: 2048 * MIB, + source: 'cgroup', + }); + }); + + it('falls back to the meminfo total only when no cgroup or explicit limit is set', () => { + expect( + computeWorkloadBudget({ + limits: [undefined], + memTotalBytes: 6 * GIB, + reserveBytes: RESERVE_BYTES, + }) + ).toEqual({ + ok: true, + containerLimitBytes: 6 * GIB, + aggregateMaxBytes: 4096 * MIB, + source: 'meminfo', + }); + expect( + computeWorkloadBudget({ limits: [], memTotalBytes: 6 * GIB, reserveBytes: RESERVE_BYTES }) + ).toEqual({ + ok: true, + containerLimitBytes: 6 * GIB, + aggregateMaxBytes: 4096 * MIB, + source: 'meminfo', + }); + expect( + computeWorkloadBudget({ + limits: [undefined], + explicitLimitBytes: 4 * GIB, + memTotalBytes: 8 * GIB, + reserveBytes: RESERVE_BYTES, + }) + ).toEqual({ + ok: true, + containerLimitBytes: 4 * GIB, + aggregateMaxBytes: 2048 * MIB, + source: 'explicit', + }); + }); + + it('reports no_finite_limit without a cgroup, explicit or meminfo limit', () => { + expect(computeWorkloadBudget({ limits: [undefined], reserveBytes: RESERVE_BYTES })).toEqual({ + ok: false, + failure: 'no_finite_limit', + }); + }); +}); + +describe('workload probe', () => { + it('uses the nearest ancestor that delegates memory by token match', () => { + const root = makeRoot(); + delegationTree(root); + const workload = initialize(root, '/a/b'); + expect(workload.enabled).toBe(true); + expect(workload.failure).toBeUndefined(); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'a', 'kilo-workloads')); + expect(workload.placement?.aggregateMaxBytes).toBe(4 * GIB); + }); + + it('places when the delegated ancestor is finite and the root memory.max is absent', () => { + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + writeControl(path.join(root, 'a', 'kilo-workloads'), 'cgroup.events', 'populated 0'); + const workload = initialize(root, '/a/b'); + expect(workload.failure).toBeUndefined(); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'a', 'kilo-workloads')); + expect(workload.placement?.aggregateMaxBytes).toBe(4 * GIB); + }); + + it('ignores a non-token controller name and falls back to a higher delegating ancestor', () => { + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'cpuset cpu memoryfoo'); + writeControl(path.join(root, 'a'), 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + writeControl(root, 'cgroup.subtree_control', 'memory'); + writeControl(root, 'memory.max', String(6 * GIB)); + writeControl(path.join(root, 'kilo-workloads'), 'cgroup.events', 'populated 0'); + const workload = initialize(root, '/a/b'); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'kilo-workloads')); + }); + + it('falls back to not_delegated when the mount root cannot be enabled', () => { + if (process.getuid?.() === 0) return; + const root = makeRoot(); + writeControl(root, 'cgroup.subtree_control', 'cpu'); + writeControl(root, 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + chmodSync(path.join(root, 'cgroup.subtree_control'), 0o444); + const workload = initialize(root, '/a/b'); + expect(workload).toEqual({ enabled: true, failure: 'not_delegated' }); + expect(workload.placement).toBeUndefined(); + expect(admitControlWorkload(workload)).toBeUndefined(); + expect(existsSync(path.join(root, 'kilo-workloads'))).toBe(false); + }); + + it('does not rewrite the mount root when an ancestor already delegates', () => { + const root = makeRoot(); + writeControl(root, 'cgroup.subtree_control', 'cpu'); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + writeControl(path.join(root, 'a', 'kilo-workloads'), 'cgroup.events', 'populated 0'); + const workload = initialize(root, '/a/b'); + expect(workload.failure).toBeUndefined(); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'a', 'kilo-workloads')); + expect(readFileSync(path.join(root, 'cgroup.subtree_control'), 'utf8').trim()).toBe('cpu'); + }); + + it('self-enables the mount root and places under it for nested membership', () => { + const root = makeRoot(); + writeControl(root, 'cgroup.subtree_control', 'cpu'); + writeControl(root, 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + const workload = initialize(root, '/a/b'); + expect(workload.failure).toBeUndefined(); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'kilo-workloads')); + expect(workload.placement?.aggregateMaxBytes).toBe(4 * GIB); + expect(readFileSync(path.join(root, 'cgroup.subtree_control'), 'utf8').trim()).toBe('+memory'); + }); + + it('self-enables its own cgroup when membership is the mount root', () => { + const root = makeRoot(); + writeControl(root, 'cgroup.subtree_control', 'cpu'); + writeControl(root, 'memory.max', String(6 * GIB)); + const workload = initialize(root, '/'); + expect(workload.failure).toBeUndefined(); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'kilo-workloads')); + expect(readFileSync(path.join(root, 'cgroup.subtree_control'), 'utf8').trim()).toBe('+memory'); + }); + + it('rejects admission after a self-enabled placement is tampered with', () => { + const root = makeRoot(); + writeControl(root, 'cgroup.subtree_control', 'cpu'); + writeControl(root, 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + const workload = initialize(root, '/a/b'); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'kilo-workloads')); + writeFileSync(path.join(root, 'kilo-workloads', 'memory.oom.group'), '1'); + expect(admitControlWorkload(workload)).toBeUndefined(); + }); + + it('continues the walk when a nearer cgroup.subtree_control is absent', () => { + const root = makeRoot(); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + writeControl(path.join(root, 'a'), 'memory.max', String(6 * GIB)); + writeControl(root, 'cgroup.subtree_control', 'memory'); + writeControl(root, 'memory.max', String(6 * GIB)); + writeControl(path.join(root, 'kilo-workloads'), 'cgroup.events', 'populated 0'); + const workload = initialize(root, '/a/b'); + expect(workload.failure).toBeUndefined(); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'kilo-workloads')); + }); + + it('fails closed as unavailable when a nearer cgroup.subtree_control cannot be read', () => { + if (process.getuid?.() === 0) return; + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', String(6 * GIB)); + writeControl(root, 'cgroup.subtree_control', 'memory'); + writeControl(root, 'memory.max', String(6 * GIB)); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + chmodSync(path.join(root, 'a', 'cgroup.subtree_control'), 0o000); + const workload = initialize(root, '/a/b'); + expect(workload).toEqual({ enabled: true, failure: 'unavailable' }); + expect(workload.placement).toBeUndefined(); + }); + + it('fails closed as no_finite_limit when the delegated ancestor is unbounded', () => { + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', 'max'); + writeControl(root, 'memory.max', 'max'); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + const workload = initialize(root, '/a/b'); + expect(workload).toEqual({ enabled: true, failure: 'no_finite_limit' }); + }); + + it('reports no_finite_limit when every governing memory.max is missing or max', () => { + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', 'max'); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + const workload = initialize(root, '/a/b'); + expect(workload).toEqual({ enabled: true, failure: 'no_finite_limit' }); + }); + + it('uses the meminfo total when no governing memory.max is finite', () => { + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', 'max'); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + const reports: Array<{ event: string; fields: Record }> = []; + const workload = initializeControlWorkload({ + env: { CONTROL_WORKLOAD_CGROUP: '1' }, + report: (event, fields) => reports.push({ event, fields }), + cgroupRoot: root, + selfCgroupFile: selfCgroupFile(root, '/a/b'), + procRoot: procWithMemTotal(root, 6 * 1024 * 1024), + isCgroupMount: () => true, + handlePath: (_descriptor, directory) => directory, + platform: 'linux', + }); + placements.push(workload); + expect(workload.failure).toBeUndefined(); + expect(workload.placement?.containerLimitBytes).toBe(6 * GIB); + expect(workload.placement?.aggregateMaxBytes).toBe(4 * GIB); + expect(workload.placement?.limitSource).toBe('meminfo'); + const applied = reports.find(entry => entry.fields.phase === 'started'); + expect(applied?.fields.workloadLimitSource).toBe('meminfo'); + }); + + it('fails closed without a placement when no finite cgroup limit exists and meminfo is missing or too small', () => { + for (const memTotalKb of [undefined, 2560 * 1024]) { + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', 'max'); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + const workload = runInitialize({ + root, + membership: '/a/b', + env: { CONTROL_WORKLOAD_CGROUP: '1' }, + ...(memTotalKb === undefined ? {} : { procRoot: procWithMemTotal(root, memTotalKb) }), + }); + expect(workload.placement).toBeUndefined(); + expect(admitControlWorkload(workload)).toBeUndefined(); + } + }); + + it('fails closed as below_minimum when the ancestor limit is too small', () => { + const root = makeRoot(); + writeControl(path.join(root, 'a'), 'cgroup.subtree_control', 'memory'); + writeControl(path.join(root, 'a'), 'memory.max', String(2560 * MIB)); + writeControl(root, 'memory.max', 'max'); + mkdirSync(path.join(root, 'a', 'b'), { recursive: true }); + const workload = initialize(root, '/a/b'); + expect(workload).toEqual({ enabled: true, failure: 'below_minimum' }); + }); + + it('lets an explicit limit tighten the ancestor limit', () => { + const root = makeRoot(); + delegationTree(root); + writeControl(path.join(root, 'a'), 'memory.max', String(8 * GIB)); + const workload = initialize(root, '/a/b', { CONTROL_WORKLOAD_LIMIT_MB: '3072' }); + expect(workload.placement?.containerLimitBytes).toBe(3 * GIB); + expect(workload.placement?.aggregateMaxBytes).toBe(1 * GIB); + }); + + it('rejects a reused parent that is already populated', () => { + const root = makeRoot(); + delegationTree(root); + writeControl(path.join(root, 'a', 'kilo-workloads'), 'cgroup.events', 'populated 1'); + const workload = initialize(root, '/a/b'); + expect(workload).toEqual({ enabled: true, failure: 'occupied' }); + }); + + it('does not treat mocked cgroup files as proof of kernel enforcement', () => { + const root = makeRoot(); + delegationTree(root); + const reports: Array<{ event: string; fields: Record }> = []; + const workload = initializeControlWorkload({ + env: { CONTROL_WORKLOAD_CGROUP: '1' }, + report: (event, fields) => reports.push({ event, fields }), + cgroupRoot: root, + selfCgroupFile: selfCgroupFile(root, '/a/b'), + isCgroupMount: () => true, + handlePath: (_descriptor, directory) => directory, + platform: 'linux', + }); + placements.push(workload); + expect(workload.placement?.appliedReadbackBytes).toBe(4 * GIB); + const applied = reports.find(entry => entry.event === 'control.workload'); + expect(applied?.fields.phase).toBe('started'); + expect(applied?.fields.workloadPhase).toBe('applied'); + expect(applied?.fields.siblingProtection).toBe(false); + expect(admitControlWorkload(workload)).toBe(workload.placement); + }); + + it('rejects admission when the retained parent control is no longer read back', () => { + const root = makeRoot(); + delegationTree(root); + const workload = initialize(root, '/a/b'); + writeFileSync(path.join(root, 'a', 'kilo-workloads', 'memory.oom.group'), '1'); + expect(admitControlWorkload(workload)).toBeUndefined(); + }); + + it('rejects admission when a reused parent swap file is no longer zero', () => { + const root = makeRoot(); + delegationTree(root); + const workload = initialize(root, '/a/b'); + writeFileSync(path.join(root, 'a', 'kilo-workloads', 'memory.swap.max'), '1'); + expect(admitControlWorkload(workload)).toBeUndefined(); + }); + + it('fails closed without a placement when a governing memory.max cannot be read', () => { + if (process.getuid?.() === 0) return; + const root = makeRoot(); + delegationTree(root); + chmodSync(path.join(root, 'a', 'memory.max'), 0o000); + const workload = initialize(root, '/a/b'); + expect(workload).toEqual({ enabled: true, failure: 'unavailable' }); + expect(workload.placement).toBeUndefined(); + }); +}); + +describe('workload enablement', () => { + it('attempts enforcement by default on linux when the flag is absent', () => { + const root = makeRoot(); + delegationTree(root); + const workload = runInitialize({ root, membership: '/a/b', env: {} }); + expect(workload.enabled).toBe(true); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'a', 'kilo-workloads')); + }); + + it('neutralizes a stale memory.max when explicitly disabled', () => { + const root = makeRoot(); + writeControl(path.join(root, 'a', 'kilo-workloads'), 'memory.max', String(4 * GIB)); + writeControl(path.join(root, 'a', 'kilo-workloads'), 'cpu.weight', '25'); + const workload = runInitialize({ + root, + membership: '/a/b', + env: { CONTROL_WORKLOAD_CGROUP: '0' }, + }); + expect(workload).toEqual({ enabled: false }); + expect(readFileSync(path.join(root, 'a', 'kilo-workloads', 'memory.max'), 'utf8').trim()).toBe( + 'max' + ); + expect(admitControlWorkload(workload)).toBeUndefined(); + }); + + it('neutralizes a stale cap under the wrapper own cgroup when explicitly disabled', () => { + const root = makeRoot(); + const owned = path.join(root, 'a', 'b', 'kilo-workloads'); + writeControl(owned, 'memory.max', String(4 * GIB)); + const workload = runInitialize({ + root, + membership: '/a/b', + env: { CONTROL_WORKLOAD_CGROUP: '0' }, + }); + expect(workload).toEqual({ enabled: false }); + expect(readFileSync(path.join(owned, 'memory.max'), 'utf8').trim()).toBe('max'); + }); + + it('neutralizes a stale cap at the mount root when membership is the mount root', () => { + const root = makeRoot(); + const owned = path.join(root, 'kilo-workloads'); + writeControl(owned, 'memory.max', String(4 * GIB)); + const workload = runInitialize({ + root, + membership: '/', + env: { CONTROL_WORKLOAD_CGROUP: '0' }, + }); + expect(workload).toEqual({ enabled: false }); + expect(readFileSync(path.join(owned, 'memory.max'), 'utf8').trim()).toBe('max'); + }); + + it('resets nested tool limits through the opened directory without following symlinks', () => { + const root = makeRoot(); + const tools = path.join(root, 'a', 'kilo-workloads', 'tools'); + writeControl(tools, 'memory.max', String(4 * GIB)); + writeControl(tools, 'cpu.weight', '25'); + const outside = path.join(root, 'outside'); + writeControl(outside, 'memory.max', String(7 * GIB)); + symlinkSync(outside, path.join(root, 'a', 'kilo-workloads', 'escape')); + const workload = runInitialize({ + root, + membership: '/a/b', + env: { CONTROL_WORKLOAD_CGROUP: '0' }, + }); + expect(workload).toEqual({ enabled: false }); + expect(readFileSync(path.join(tools, 'memory.max'), 'utf8').trim()).toBe('max'); + expect(readFileSync(path.join(tools, 'cpu.weight'), 'utf8').trim()).toBe('100'); + expect(readFileSync(path.join(outside, 'memory.max'), 'utf8').trim()).toBe(String(7 * GIB)); + }); + + it('resets nested tool limits through the opened descriptor on linux', () => { + if (process.platform !== 'linux') return; + const root = makeRoot(); + const tools = path.join(root, 'a', 'kilo-workloads', 'tools'); + writeControl(tools, 'memory.max', String(4 * GIB)); + const workload = initializeControlWorkload({ + env: { CONTROL_WORKLOAD_CGROUP: '0' }, + cgroupRoot: root, + selfCgroupFile: selfCgroupFile(root, '/a/b'), + isCgroupMount: () => true, + handlePath: descriptor => `/proc/self/fd/${descriptor}`, + }); + placements.push(workload); + expect(workload).toEqual({ enabled: false }); + expect(readFileSync(path.join(tools, 'memory.max'), 'utf8').trim()).toBe('max'); + }); + + it('enables for any flag value other than an exact zero', () => { + for (const value of ['1', 'false', 'off']) { + const root = makeRoot(); + delegationTree(root); + const workload = runInitialize({ + root, + membership: '/a/b', + env: { CONTROL_WORKLOAD_CGROUP: value }, + }); + expect(workload.placement?.parentDirectory).toBe(path.join(root, 'a', 'kilo-workloads')); + } + }); + + it('fails closed as unavailable without a placement off linux even by default', () => { + const root = makeRoot(); + const workload = runInitialize({ root, membership: '/a/b', env: {}, platform: 'darwin' }); + expect(workload).toEqual({ enabled: true, failure: 'unavailable' }); + expect(workload.placement).toBeUndefined(); + }); + + it('starts unmanaged for an enabled-by-default failure result from initialization', () => { + const root = makeRoot(); + const workload = runInitialize({ root, membership: '/a/b', env: {}, platform: 'darwin' }); + expect(workload.failure).toBe('unavailable'); + expect(admitControlWorkload(workload)).toBeUndefined(); + }); +}); + +describe('workload admission', () => { + it('returns undefined for an absent or disabled workload without verifying', () => { + expect(admitControlWorkload(undefined)).toBeUndefined(); + expect(admitControlWorkload({ enabled: false })).toBeUndefined(); + }); + + it('returns undefined for an enabled workload without a placement', () => { + expect(admitControlWorkload({ enabled: true })).toBeUndefined(); + expect(admitControlWorkload({ enabled: true, failure: 'unavailable' })).toBeUndefined(); + }); + + it('starts unmanaged when not_delegated has no placement', () => { + expect(admitControlWorkload({ enabled: true, failure: 'not_delegated' })).toBeUndefined(); + }); + + it('emits a failed diagnostic and returns undefined when retained admission verification fails', () => { + const root = makeRoot(); + delegationTree(root); + const reports: Array<{ event: string; fields: Record }> = []; + const workload = initializeControlWorkload({ + env: { CONTROL_WORKLOAD_CGROUP: '1' }, + report: (event, fields) => reports.push({ event, fields }), + cgroupRoot: root, + selfCgroupFile: selfCgroupFile(root, '/a/b'), + isCgroupMount: () => true, + handlePath: (_descriptor, directory) => directory, + platform: 'linux', + }); + placements.push(workload); + reports.length = 0; + writeFileSync(path.join(root, 'a', 'kilo-workloads', 'memory.oom.group'), '1'); + expect(admitControlWorkload(workload)).toBeUndefined(); + const failed = reports.find( + entry => entry.event === 'control.workload' && entry.fields.phase === 'failed' + ); + expect(failed?.fields.workloadPhase).toBe('failed'); + expect(failed?.fields.workloadFailure).toBe('readback_mismatch'); + }); + + it('starts unmanaged and reports failure when managed containment is unavailable', () => { + const root = makeRoot(); + const directory = path.join(root, 'parent'); + mkdirSync(directory, { recursive: true }); + const parentFd = openSync(directory, constants.O_RDONLY | constants.O_DIRECTORY); + const stat = fstatSync(parentFd); + const reports: Array<{ event: string; fields: Record }> = []; + const placement: WorkloadPlacement = { + parentFd, + parentDev: stat.dev, + parentIno: stat.ino, + parentReference: directory, + parentDirectory: directory, + aggregateMaxBytes: 1 * GIB, + appliedReadbackBytes: 1 * GIB, + containerLimitBytes: 3 * GIB, + limitSource: 'cgroup', + cpuWeight: 50, + cpuController: false, + sweepIntervalMs: 1000, + report: (event, fields) => reports.push({ event, fields }), + }; + let childPid: number | undefined; + try { + const scope = createOwnedProcessScope(placement); + const child = scope.spawn(process.execPath, ['-e', 'process.exit(0)'], { + cwd: process.cwd(), + env: process.env, + }); + childPid = child.pid; + expect(child.pid).toBeDefined(); + expect(scope.observesOccupancy()).toBe(false); + const failed = reports.find( + entry => entry.event === 'control.workload' && entry.fields.phase === 'failed' + ); + expect(failed?.fields.workloadPhase).toBe('failed'); + expect(failed?.fields.workloadFailure).toBe('unavailable'); + } finally { + if (childPid !== undefined) { + try { + process.kill(childPid, 'SIGKILL'); + } catch { + childPid = undefined; + } + } + closeSync(parentFd); + } + }); +}); + +describe('workload classification', () => { + it('classifies the server chain, tool commands, orphans and non-members', () => { + const table = new Map([ + [100, { pid: 100, ppid: 1, argv: ['/bin/sh', '-c', 'exec kilo serve'] }], + [101, { pid: 101, ppid: 100, argv: ['/usr/local/bin/kilo', 'serve', '--port=0'] }], + [102, { pid: 102, ppid: 100, argv: ['bash', '-c', 'npm run build'] }], + [103, { pid: 103, ppid: 101, argv: ['node', 'node_modules/.bin/tsc'] }], + [104, { pid: 104, ppid: 1, argv: ['bash', '-c', 'orphaned build'] }], + [105, { pid: 105, ppid: 100, argv: ['unrelated'] }], + ]); + const members = [100, 101, 102, 103, 104]; + const { serverPids, toolPids } = classifyWorkloadMembers(members, table, 100); + expect(serverPids).toEqual([101]); + expect(new Set(toolPids)).toEqual(new Set([102, 103, 104])); + expect(toolPids).not.toContain(105); + }); +}); + +describe('managed workload limits', () => { + function managedTree(): { parent: string; server: string; tools: string } { + const root = makeRoot(); + const parent = path.join(root, 'scope'); + const server = path.join(parent, 'server'); + const tools = path.join(parent, 'tools'); + mkdirSync(server, { recursive: true }); + mkdirSync(tools, { recursive: true }); + return { parent, server, tools }; + } + + const apply = (tree: { parent: string; server: string; tools: string }) => + applyManagedWorkloadLimits({ + parentReference: tree.parent, + serverReference: tree.server, + toolsReference: tree.tools, + aggregateMaxBytes: 4 * GIB, + }); + + it('applies the required limits and reports the cpu read-back', () => { + const tree = managedTree(); + expect(apply(tree)).toEqual({ cpuController: true }); + expect(readFileSync(path.join(tree.tools, 'memory.max'), 'utf8')).toBe(String(4 * GIB)); + expect(readFileSync(path.join(tree.tools, 'memory.oom.group'), 'utf8')).toBe('1'); + expect(readFileSync(path.join(tree.server, 'memory.oom.group'), 'utf8')).toBe('0'); + }); + + it('treats a missing swap file and cpu file as non-fatal', () => { + if (process.getuid?.() === 0) return; + const tree = managedTree(); + writeFileSync(path.join(tree.tools, 'memory.max'), String(4 * GIB)); + writeFileSync(path.join(tree.tools, 'memory.oom.group'), '1'); + writeFileSync(path.join(tree.server, 'memory.oom.group'), '0'); + chmodSync(tree.tools, 0o555); + try { + expect(apply(tree)).toEqual({ cpuController: false }); + } finally { + chmodSync(tree.tools, 0o755); + } + }); + + it('fails admission when a required tools read-back disagrees', () => { + if (process.getuid?.() === 0) return; + const tree = managedTree(); + writeFileSync(path.join(tree.tools, 'memory.max'), String(2 * GIB)); + chmodSync(path.join(tree.tools, 'memory.max'), 0o444); + try { + expect(() => apply(tree)).toThrow(); + } finally { + chmodSync(path.join(tree.tools, 'memory.max'), 0o644); + } + }); + + it('fails admission when an existing tools swap file is not zero', () => { + if (process.getuid?.() === 0) return; + const tree = managedTree(); + writeFileSync(path.join(tree.tools, 'memory.swap.max'), '1'); + chmodSync(path.join(tree.tools, 'memory.swap.max'), 0o444); + try { + expect(() => apply(tree)).toThrow(); + } finally { + chmodSync(path.join(tree.tools, 'memory.swap.max'), 0o644); + } + }); +}); + +describe('workload pressure parsing', () => { + it('parses integer total counters only', () => { + const text = + 'some avg10=0.00 avg60=0.00 avg300=0.00 total=123\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=45'; + expect(parsePressureTotal(text, 'some')).toBe(123); + expect(parsePressureTotal(text, 'full')).toBe(45); + }); + + it('omits missing or unparsable pressure', () => { + expect(parsePressureTotal(undefined, 'some')).toBeUndefined(); + expect(parsePressureTotal('garbage', 'some')).toBeUndefined(); + expect(parsePressureTotal('some avg10=0.00 total=abc', 'some')).toBeUndefined(); + }); +}); + +describe('workload diagnostics', () => { + it('round-trips through the helper and omits unknown fields rejected by the strict schema', () => { + const record = createControlDiagnosticRecord( + 'control.workload', + { + phase: 'started', + workloadPhase: 'applied', + aggregateMaxBytes: 2048 * MIB, + workloadLimitSource: 'meminfo', + siblingProtection: false, + unknownWorkloadField: 'secret', + }, + Date.now() + ); + expect(record?.event).toBe('control.workload'); + expect(record?.fields.phase).toBe('started'); + expect(record?.fields.workloadPhase).toBe('applied'); + expect(record?.fields.workloadLimitSource).toBe('meminfo'); + expect(record?.fields).not.toHaveProperty('unknownWorkloadField'); + expect( + controlDiagnosticRecordSchema.safeParse({ + timestamp: Date.now(), + event: 'control.workload', + fields: { + phase: 'started', + workloadPhase: 'applied', + unknownWorkloadField: 'secret', + }, + }).success + ).toBe(false); + }); + + it('emits enabled early failures and stays silent when the flag is zero', () => { + const reports: Array<{ event: string; fields: Record }> = []; + const report = (event: string, fields: Record): void => { + reports.push({ event, fields }); + }; + const disabled = initializeControlWorkload({ + env: { CONTROL_WORKLOAD_CGROUP: '0' }, + platform: 'darwin', + report, + }); + expect(disabled).toEqual({ enabled: false }); + expect(reports).toEqual([]); + + const enabled = initializeControlWorkload({ env: {}, platform: 'darwin', report }); + expect(enabled).toEqual({ enabled: true, failure: 'unavailable' }); + expect(reports).toHaveLength(1); + expect(reports[0].event).toBe('control.workload'); + expect(reports[0].fields.phase).toBe('failed'); + expect(reports[0].fields.workloadPhase).toBe('failed'); + expect(reports[0].fields.workloadFailure).toBe('unavailable'); + }); + + it('deduplicates identical failures but re-emits when oom counts change', () => { + const reports: Array<{ event: string; fields: Record }> = []; + const reporter = createWorkloadReporter((event, fields) => reports.push({ event, fields })); + reporter.emit('kilo-control-1', { + phase: 'failed', + workloadPhase: 'migration', + workloadFailure: 'pid_changed', + }); + reporter.emit('kilo-control-1', { + phase: 'failed', + workloadPhase: 'migration', + workloadFailure: 'pid_changed', + }); + reporter.emit('kilo-control-1', { + phase: 'failed', + workloadPhase: 'oom', + oomKills: 1, + oomGroupKills: 0, + }); + reporter.emit('kilo-control-1', { + phase: 'failed', + workloadPhase: 'oom', + oomKills: 1, + oomGroupKills: 0, + }); + reporter.emit('kilo-control-1', { + phase: 'failed', + workloadPhase: 'oom', + oomKills: 2, + oomGroupKills: 0, + }); + expect(reports).toHaveLength(3); + expect(reports[0].fields.workloadFailure).toBe('pid_changed'); + expect(reports[1].fields.oomKills).toBe(1); + expect(reports[2].fields.oomKills).toBe(2); + }); + + it('re-emits changed memory stats and keeps identical stats deduplicated', () => { + const reports: Array<{ event: string; fields: Record }> = []; + const reporter = createWorkloadReporter((event, fields) => reports.push({ event, fields })); + const emit = (currentBytes: number): void => { + reporter.emit('kilo-control-1', { + phase: 'completed', + workloadPhase: 'stats', + currentBytes, + }); + }; + emit(1024); + emit(1024); + emit(2048); + expect(reports).toHaveLength(2); + expect(reports[0].fields.currentBytes).toBe(1024); + expect(reports[1].fields.currentBytes).toBe(2048); + reporter.emit('kilo-control-1', { + phase: 'completed', + workloadPhase: 'stats', + currentBytes: 2048, + toolCount: 1, + serverCount: 1, + migratedCount: 0, + cpuController: true, + }); + reporter.emit('kilo-control-1', { + phase: 'completed', + workloadPhase: 'stats', + currentBytes: 2048, + toolCount: 2, + serverCount: 1, + migratedCount: 0, + cpuController: true, + }); + expect(reports).toHaveLength(4); + expect(reports[3].fields.toolCount).toBe(2); + }); +}); diff --git a/services/cloud-agent-next/wrapper/src/control/workload-cgroup.ts b/services/cloud-agent-next/wrapper/src/control/workload-cgroup.ts new file mode 100644 index 0000000000..599050c70b --- /dev/null +++ b/services/cloud-agent-next/wrapper/src/control/workload-cgroup.ts @@ -0,0 +1,877 @@ +import { spawnSync } from 'node:child_process'; +import { + closeSync, + constants, + fstatSync, + lstatSync, + mkdirSync, + openSync, + readdirSync, + readFileSync, + statfsSync, + writeFileSync, +} from 'node:fs'; +import path from 'node:path'; +import type { + ControlDiagnosticFields, + ControlDiagnosticReporter, +} from '../../../src/shared/control-diagnostics.js'; +import { isKiloServerProcess, readMemTotalBytes } from '../tool-cgroup.js'; + +export const CONTROL_WORKLOAD_CGROUP_ENV = 'CONTROL_WORKLOAD_CGROUP'; +export const CONTROL_WORKLOAD_RESERVE_MB_ENV = 'CONTROL_WORKLOAD_RESERVE_MB'; +export const CONTROL_WORKLOAD_LIMIT_MB_ENV = 'CONTROL_WORKLOAD_LIMIT_MB'; + +export const DEFAULT_CONTROL_RESERVE_BYTES = 2048 * 1024 * 1024; +export const MIN_WORKLOAD_CAP_BYTES = 1024 * 1024 * 1024; +export const WORKLOAD_CPU_WEIGHT = 50; +export const WORKLOAD_SWEEP_INTERVAL_MS = 1000; +export const WORKLOAD_PARENT_NAME = 'kilo-workloads'; +export const WORKLOAD_SERVER_NAME = 'server'; +export const WORKLOAD_TOOLS_NAME = 'tools'; +export const WORKLOAD_DEFAULT_CPU_WEIGHT = 100; +export const CGROUP_FS_MAGIC = 0x63677270; + +export type WorkloadFailure = + | 'flag_off' + | 'no_finite_limit' + | 'below_minimum' + | 'not_delegated' + | 'occupied' + | 'controller_unavailable' + | 'readback_mismatch' + | 'write_failed' + | 'unavailable' + | 'pid_changed' + | 'membership_unconfirmed'; + +export type WorkloadPhase = + | 'probe' + | 'applied' + | 'migration' + | 'oom' + | 'stats' + | 'rollback' + | 'failed'; + +export type WorkloadLimitSource = 'cgroup' | 'explicit' | 'meminfo'; + +export type WorkloadPlacement = { + parentFd: number; + parentDev: number; + parentIno: number; + parentReference: string; + parentDirectory: string; + aggregateMaxBytes: number; + appliedReadbackBytes: number; + containerLimitBytes: number; + limitSource: WorkloadLimitSource; + cpuWeight: number; + cpuController: boolean; + sweepIntervalMs: number; + report?: ControlDiagnosticReporter; +}; + +export type ControlWorkload = { + enabled: boolean; + placement?: WorkloadPlacement; + failure?: WorkloadFailure; +}; + +export type WorkloadProcessEntry = { pid: number; ppid: number; argv: string[] }; + +export type WorkloadStats = { + currentBytes?: number; + peakBytes?: number; + oomKills: number; + oomGroupKills: number; + pressureSomeTotal?: number; + pressureFullTotal?: number; +}; + +export class WorkloadUnavailableError extends Error { + constructor(readonly failure: WorkloadFailure) { + super(`Control workload unavailable: ${failure}`); + this.name = 'WorkloadUnavailableError'; + } +} + +type WorkloadEmission = ControlDiagnosticFields & { + phase: 'started' | 'completed' | 'failed'; + workloadPhase: WorkloadPhase; + workloadFailure?: WorkloadFailure; +}; + +export type WorkloadReporter = { + emit(scopeId: string | undefined, fields: WorkloadEmission): void; +}; + +export type ControlWorkloadOptions = { + env: Record; + report?: ControlDiagnosticReporter; + cgroupRoot?: string; + selfCgroupFile?: string; + procRoot?: string; + isCgroupMount?: (root: string) => boolean; + handlePath?: (descriptor: number, directory: string) => string; + platform?: NodeJS.Platform; +}; + +type ProbeResult = + | { ok: true; placement: WorkloadPlacement } + | { ok: false; failure: WorkloadFailure }; + +export function createWorkloadReporter(report?: ControlDiagnosticReporter): WorkloadReporter { + const last = new Map(); + return { + emit(scopeId, fields) { + if (!report) return; + const key = [ + fields.phase, + fields.workloadPhase, + fields.workloadFailure ?? '', + fields.oomKills ?? '', + fields.oomGroupKills ?? '', + fields.currentBytes ?? '', + fields.peakBytes ?? '', + fields.pressureSomeTotal ?? '', + fields.pressureFullTotal ?? '', + fields.toolCount ?? '', + fields.serverCount ?? '', + fields.migratedCount ?? '', + fields.cpuController ?? '', + ].join(':'); + const bucket = scopeId ?? 'global'; + if (last.get(bucket) === key) return; + last.set(bucket, key); + try { + report('control.workload', scopeId ? { ...fields, scopeId } : { ...fields }); + } catch { + return; + } + }, + }; +} + +function parsePositiveMb(value: string | undefined): number | undefined { + if (value === undefined) return undefined; + const parsed = Number.parseInt(value, 10); + return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : undefined; +} + +export function computeWorkloadBudget(input: { + limits: (number | undefined)[]; + explicitLimitBytes?: number; + memTotalBytes?: number; + reserveBytes: number; +}): + | { + ok: true; + containerLimitBytes: number; + aggregateMaxBytes: number; + source: WorkloadLimitSource; + } + | { ok: false; failure: WorkloadFailure } { + const finite = input.limits.filter( + (value): value is number => typeof value === 'number' && Number.isFinite(value) && value > 0 + ); + let limit: number; + let source: WorkloadLimitSource; + if (finite.length > 0) { + limit = Math.min(...finite); + source = 'cgroup'; + if (input.explicitLimitBytes !== undefined && input.explicitLimitBytes < limit) { + limit = input.explicitLimitBytes; + source = 'explicit'; + } + } else if (input.explicitLimitBytes !== undefined) { + limit = input.explicitLimitBytes; + source = 'explicit'; + } else if ( + typeof input.memTotalBytes === 'number' && + Number.isFinite(input.memTotalBytes) && + input.memTotalBytes > 0 + ) { + limit = input.memTotalBytes; + source = 'meminfo'; + } else { + return { ok: false, failure: 'no_finite_limit' }; + } + const aggregateMaxBytes = limit - input.reserveBytes; + if (aggregateMaxBytes < MIN_WORKLOAD_CAP_BYTES) { + return { ok: false, failure: 'below_minimum' }; + } + return { ok: true, containerLimitBytes: limit, aggregateMaxBytes, source }; +} + +export function classifyWorkloadMembers( + members: Iterable, + table: Map, + rootPid: number +): { serverPids: number[]; toolPids: number[] } { + const memberSet = new Set(members); + const childrenByPpid = new Map(); + for (const entry of table.values()) { + const siblings = childrenByPpid.get(entry.ppid); + if (siblings) siblings.push(entry); + else childrenByPpid.set(entry.ppid, [entry]); + } + const serverPids: number[] = []; + const toolPids: number[] = []; + const reached = new Set(); + const queue = [rootPid]; + for (let index = 0; index < queue.length; index += 1) { + for (const child of childrenByPpid.get(queue[index]) ?? []) { + if (!memberSet.has(child.pid) || reached.has(child.pid)) continue; + reached.add(child.pid); + (isKiloServerProcess(child.argv) ? serverPids : toolPids).push(child.pid); + queue.push(child.pid); + } + } + for (const pid of memberSet) { + if (pid === rootPid || reached.has(pid)) continue; + toolPids.push(pid); + } + return { serverPids, toolPids }; +} + +export function parsePressureTotal( + text: string | undefined, + kind: 'some' | 'full' +): number | undefined { + if (text === undefined) return undefined; + for (const line of text.split('\n')) { + if (!line.startsWith(`${kind} `)) continue; + const match = /(?:^|\s)total=(\d+)(?:\s|$)/.exec(line); + if (!match) return undefined; + const parsed = Number.parseInt(match[1], 10); + return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : undefined; + } + return undefined; +} + +export function readWorkloadStats(reference: string): WorkloadStats { + const stats: WorkloadStats = { oomKills: 0, oomGroupKills: 0 }; + const current = readControl(path.join(reference, 'memory.current')); + const peak = readControl(path.join(reference, 'memory.peak')); + const pressure = readControl(path.join(reference, 'memory.pressure')); + const events = readControl(path.join(reference, 'memory.events')); + if (current.ok) { + const parsed = Number.parseInt(current.text.trim(), 10); + if (Number.isSafeInteger(parsed) && parsed >= 0) stats.currentBytes = parsed; + } + if (peak.ok) { + const parsed = Number.parseInt(peak.text.trim(), 10); + if (Number.isSafeInteger(parsed) && parsed >= 0) stats.peakBytes = parsed; + } + const some = parsePressureTotal(pressure.ok ? pressure.text : undefined, 'some'); + const full = parsePressureTotal(pressure.ok ? pressure.text : undefined, 'full'); + if (some !== undefined) stats.pressureSomeTotal = some; + if (full !== undefined) stats.pressureFullTotal = full; + if (events.ok) { + for (const line of events.text.split('\n')) { + const [key, value] = line.trim().split(/\s+/); + const parsed = Number.parseInt(value ?? '', 10); + if (!Number.isSafeInteger(parsed) || parsed < 0) continue; + if (key === 'oom_kill') stats.oomKills = parsed; + if (key === 'oom_group_kill') stats.oomGroupKills = parsed; + } + } + return stats; +} + +export function readWorkloadMembership(selfCgroupFile: string): string | undefined { + const read = readControl(selfCgroupFile); + if (!read.ok) return undefined; + const line = read.text.split('\n').find(candidate => candidate.startsWith('0::')); + const membership = line?.slice(3); + if ( + !membership || + !path.posix.isAbsolute(membership) || + path.posix.normalize(membership) !== membership + ) { + return undefined; + } + return membership; +} + +export function workloadAncestors(membership: string): string[] { + const ancestors: string[] = []; + let current = path.posix.dirname(membership); + while (current !== membership) { + ancestors.push(current); + const next = path.posix.dirname(current); + if (next === current) break; + current = next; + } + return ancestors; +} + +type ControlRead = { ok: true; text: string } | { ok: false; missing: boolean }; + +function isEnoent(error: unknown): boolean { + return typeof error === 'object' && error !== null && 'code' in error && error.code === 'ENOENT'; +} + +function readControl(pathname: string): ControlRead { + try { + return { ok: true, text: readFileSync(pathname, 'utf8') }; + } catch (error) { + return { ok: false, missing: isEnoent(error) }; + } +} + +function writeControlOptional(pathname: string, value: string): void { + try { + writeFileSync(pathname, value); + } catch { + return; + } +} + +type TokenSetRead = { ok: true; tokens: Set } | { ok: false; missing: boolean }; + +function readTokenSet(pathname: string): TokenSetRead { + const read = readControl(pathname); + if (!read.ok) return { ok: false, missing: read.missing }; + return { + ok: true, + tokens: new Set( + read.text + .trim() + .split(/\s+/) + .filter(Boolean) + .map(token => token.replace(/^[+-]/, '')) + ), + }; +} + +type MemoryMaxRead = + | { kind: 'limit'; bytes: number } + | { kind: 'unbounded' } + | { kind: 'missing' } + | { kind: 'unreadable' }; + +function readMemoryMax(pathname: string): MemoryMaxRead { + const read = readControl(pathname); + if (!read.ok) return read.missing ? { kind: 'missing' } : { kind: 'unreadable' }; + const trimmed = read.text.trim(); + if (trimmed === 'max') return { kind: 'unbounded' }; + const parsed = Number(trimmed); + return Number.isSafeInteger(parsed) && parsed > 0 + ? { kind: 'limit', bytes: parsed } + : { kind: 'unreadable' }; +} + +function parsePopulation(pathname: string): number | undefined { + const read = readControl(pathname); + if (!read.ok) return undefined; + const entries = read.text.split('\n').filter(line => line.startsWith('populated ')); + if (entries.length !== 1 || !/^populated [01]$/.test(entries[0])) return undefined; + return Number(entries[0].slice('populated '.length)); +} + +function isErofs(error: unknown): boolean { + return typeof error === 'object' && error !== null && 'code' in error && error.code === 'EROFS'; +} + +function remountCgroupWritable(target: string): void { + spawnSync('mount', ['-o', 'remount,rw', target], { stdio: 'ignore' }); +} + +function defaultIsCgroupMount(root: string): boolean { + try { + return statfsSync(root).type === CGROUP_FS_MAGIC; + } catch { + return false; + } +} + +function defaultHandlePath(descriptor: number): string { + return `/proc/self/fd/${descriptor}`; +} + +type AncestorLookup = + | { ok: true; relative: string; directory: string } + | { ok: false; failure: 'not_delegated' | 'unavailable' }; + +function workloadCgroupCandidates(membership: string): string[] { + return [membership, ...workloadAncestors(membership)]; +} + +function candidateDirectory(cgroupRoot: string, relative: string): string { + return relative === '/' ? cgroupRoot : path.join(cgroupRoot, relative); +} + +function scanDelegatingCgroup(cgroupRoot: string, candidates: string[]): AncestorLookup { + for (const relative of candidates) { + const directory = candidateDirectory(cgroupRoot, relative); + const control = readTokenSet(path.join(directory, 'cgroup.subtree_control')); + if (!control.ok) { + if (control.missing) continue; + return { ok: false, failure: 'unavailable' }; + } + if (control.tokens.has('memory')) return { ok: true, relative, directory }; + } + return { ok: false, failure: 'not_delegated' }; +} + +function enableMountRootControllers(cgroupRoot: string): void { + const controlPath = path.join(cgroupRoot, 'cgroup.subtree_control'); + try { + const enabled = new Set(readFileSync(controlPath, 'utf8').trim().split(/\s+/).filter(Boolean)); + const missing = ['memory', 'cpu'].filter(controller => !enabled.has(controller)); + if (missing.length > 0) { + writeFileSync(controlPath, missing.map(controller => `+${controller}`).join(' ')); + } + } catch { + return; + } +} + +function findUsableParent(cgroupRoot: string, membership: string): AncestorLookup { + const delegatedAncestor = scanDelegatingCgroup(cgroupRoot, workloadAncestors(membership)); + if (delegatedAncestor.ok || delegatedAncestor.failure === 'unavailable') return delegatedAncestor; + enableMountRootControllers(cgroupRoot); + return scanDelegatingCgroup(cgroupRoot, workloadCgroupCandidates(membership)); +} + +function probeWorkloadParent(input: { + usable: { relative: string; directory: string }; + budget: { containerLimitBytes: number; aggregateMaxBytes: number; source: WorkloadLimitSource }; + handlePath: (descriptor: number, directory: string) => string; + report?: ControlDiagnosticReporter; +}): ProbeResult { + const parentDirectory = path.join(input.usable.directory, WORKLOAD_PARENT_NAME); + let parentFd: number | undefined; + let retained = false; + try { + let exists = false; + try { + const stat = lstatSync(parentDirectory); + if (stat.isSymbolicLink() || !stat.isDirectory()) return { ok: false, failure: 'occupied' }; + exists = true; + } catch { + exists = false; + } + if (exists) { + const populated = parsePopulation(path.join(parentDirectory, 'cgroup.events')); + if (populated !== 0) return { ok: false, failure: 'occupied' }; + } else { + try { + mkdirSync(parentDirectory); + } catch (error) { + if (!isErofs(error)) return { ok: false, failure: 'write_failed' }; + remountCgroupWritable(input.usable.directory); + try { + mkdirSync(parentDirectory); + } catch { + return { ok: false, failure: 'write_failed' }; + } + } + } + try { + parentFd = openSync( + parentDirectory, + constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW + ); + } catch { + return { ok: false, failure: 'unavailable' }; + } + const reference = input.handlePath(parentFd, parentDirectory); + const { dev, ino } = fstatSync(parentFd); + + try { + writeFileSync(path.join(reference, 'memory.max'), String(input.budget.aggregateMaxBytes)); + } catch { + return { ok: false, failure: 'write_failed' }; + } + const applied = readMemoryMax(path.join(reference, 'memory.max')); + if (applied.kind !== 'limit' || applied.bytes !== input.budget.aggregateMaxBytes) { + return { ok: false, failure: 'readback_mismatch' }; + } + const appliedReadbackBytes = applied.bytes; + try { + writeFileSync(path.join(reference, 'memory.oom.group'), '0'); + } catch { + return { ok: false, failure: 'write_failed' }; + } + const oomGroup = readControl(path.join(reference, 'memory.oom.group')); + if (!oomGroup.ok || oomGroup.text.trim() !== '0') { + return { ok: false, failure: 'readback_mismatch' }; + } + try { + writeFileSync(path.join(reference, 'memory.swap.max'), '0'); + } catch { + // memory.swap.max is optional on kernels without swap accounting. + } + const swap = readControl(path.join(reference, 'memory.swap.max')); + if (swap.ok ? swap.text.trim() !== '0' : !swap.missing) { + return { ok: false, failure: 'readback_mismatch' }; + } + + let cpuController = false; + try { + writeFileSync(path.join(reference, 'cpu.weight'), String(WORKLOAD_CPU_WEIGHT)); + const cpu = readControl(path.join(reference, 'cpu.weight')); + cpuController = cpu.ok && cpu.text.trim() === String(WORKLOAD_CPU_WEIGHT); + } catch { + cpuController = false; + } + + try { + writeFileSync(path.join(reference, 'cgroup.subtree_control'), '+memory +cpu'); + } catch { + try { + writeFileSync(path.join(reference, 'cgroup.subtree_control'), '+memory'); + } catch { + return { ok: false, failure: 'controller_unavailable' }; + } + } + const subtree = readTokenSet(path.join(reference, 'cgroup.subtree_control')); + if (!subtree.ok || !subtree.tokens.has('memory')) { + return { ok: false, failure: 'controller_unavailable' }; + } + + const placement: WorkloadPlacement = { + parentFd, + parentDev: dev, + parentIno: ino, + parentReference: reference, + parentDirectory, + aggregateMaxBytes: input.budget.aggregateMaxBytes, + appliedReadbackBytes, + containerLimitBytes: input.budget.containerLimitBytes, + limitSource: input.budget.source, + cpuWeight: WORKLOAD_CPU_WEIGHT, + cpuController, + sweepIntervalMs: WORKLOAD_SWEEP_INTERVAL_MS, + ...(input.report ? { report: input.report } : {}), + }; + retained = true; + return { ok: true, placement }; + } catch { + return { ok: false, failure: 'unavailable' }; + } finally { + if (!retained && parentFd !== undefined) { + try { + closeSync(parentFd); + } catch { + // Descriptor close failures are not fatal here. + } + } + } +} + +function resetWorkloadDirectory(reference: string): boolean { + let ok = true; + try { + const current = readControl(path.join(reference, 'memory.max')); + if (!current.ok || current.text.trim() !== 'max') { + writeFileSync(path.join(reference, 'memory.max'), 'max'); + } + } catch { + ok = false; + } + try { + const current = readControl(path.join(reference, 'cpu.weight')); + if (!current.ok || current.text.trim() !== String(WORKLOAD_DEFAULT_CPU_WEIGHT)) { + writeFileSync(path.join(reference, 'cpu.weight'), String(WORKLOAD_DEFAULT_CPU_WEIGHT)); + } + } catch { + ok = false; + } + return ok; +} + +function resetWorkloadTree( + directory: string, + depth: number, + handlePath: (descriptor: number, directory: string) => string +): boolean { + let descriptor: number | undefined; + try { + descriptor = openSync( + directory, + constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW + ); + } catch { + return false; + } + try { + const reference = handlePath(descriptor, directory); + let ok = resetWorkloadDirectory(reference); + if (depth <= 0) return ok; + let entries; + try { + entries = readdirSync(reference, { withFileTypes: true }); + } catch { + return false; + } + for (const entry of entries) { + if (entry.isSymbolicLink() || !entry.isDirectory()) continue; + if (!resetWorkloadTree(path.join(reference, entry.name), depth - 1, handlePath)) ok = false; + } + return ok; + } finally { + try { + closeSync(descriptor); + } catch { + // Descriptor close failures are not fatal here. + } + } +} + +function neutralizeControlWorkload(input: { + cgroupRoot: string; + membership: string; + handlePath: (descriptor: number, directory: string) => string; + report?: ControlDiagnosticReporter; +}): void { + const reporter = createWorkloadReporter(input.report); + for (const relative of workloadCgroupCandidates(input.membership)) { + const candidate = path.join( + candidateDirectory(input.cgroupRoot, relative), + WORKLOAD_PARENT_NAME + ); + let stat; + try { + stat = lstatSync(candidate); + } catch { + continue; + } + if (stat.isSymbolicLink() || !stat.isDirectory()) continue; + const ok = resetWorkloadTree(candidate, 2, input.handlePath); + reporter.emit(undefined, { + phase: ok ? 'completed' : 'failed', + workloadPhase: 'rollback', + ...(ok ? {} : { workloadFailure: 'write_failed' as const }), + }); + } +} + +export function initializeControlWorkload(options: ControlWorkloadOptions): ControlWorkload { + const env = options.env; + const platform = options.platform ?? process.platform; + const enabled = env[CONTROL_WORKLOAD_CGROUP_ENV] !== '0'; + const reporter = createWorkloadReporter(options.report); + if (platform !== 'linux') { + if (!enabled) return { enabled: false }; + reporter.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: 'unavailable', + }); + return { enabled: true, failure: 'unavailable' }; + } + + const cgroupRoot = options.cgroupRoot ?? '/sys/fs/cgroup'; + const selfCgroupFile = options.selfCgroupFile ?? '/proc/self/cgroup'; + const procRoot = options.procRoot ?? '/proc'; + const isCgroupMount = options.isCgroupMount ?? defaultIsCgroupMount; + const handlePath = options.handlePath ?? defaultHandlePath; + const membership = readWorkloadMembership(selfCgroupFile); + if (membership === undefined) { + if (!enabled) return { enabled: false }; + reporter.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: 'unavailable', + }); + return { enabled: true, failure: 'unavailable' }; + } + + if (!enabled) { + if (isCgroupMount(cgroupRoot)) { + neutralizeControlWorkload({ + cgroupRoot, + membership, + handlePath, + ...(options.report ? { report: options.report } : {}), + }); + } + return { enabled: false }; + } + + if (!isCgroupMount(cgroupRoot)) { + reporter.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: 'unavailable', + }); + return { enabled: true, failure: 'unavailable' }; + } + + const usable = findUsableParent(cgroupRoot, membership); + if (!usable.ok) { + reporter.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: usable.failure, + }); + return { enabled: true, failure: usable.failure }; + } + + const budgetDirectories = [ + usable.directory, + ...workloadAncestors(usable.relative).map(ancestor => path.join(cgroupRoot, ancestor)), + ]; + const limits: (number | undefined)[] = []; + for (const directory of budgetDirectories) { + const read = readMemoryMax(path.join(directory, 'memory.max')); + if (read.kind === 'unreadable') { + reporter.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: 'unavailable', + }); + return { enabled: true, failure: 'unavailable' }; + } + limits.push(read.kind === 'limit' ? read.bytes : undefined); + } + const budget = computeWorkloadBudget({ + limits, + explicitLimitBytes: (() => { + const mb = parsePositiveMb(env[CONTROL_WORKLOAD_LIMIT_MB_ENV]); + return mb === undefined ? undefined : mb * 1024 * 1024; + })(), + memTotalBytes: readMemTotalBytes(procRoot), + reserveBytes: (() => { + const mb = parsePositiveMb(env[CONTROL_WORKLOAD_RESERVE_MB_ENV]); + return mb === undefined ? DEFAULT_CONTROL_RESERVE_BYTES : mb * 1024 * 1024; + })(), + }); + if (!budget.ok) { + reporter.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: budget.failure, + }); + return { enabled: true, failure: budget.failure }; + } + + const probe = probeWorkloadParent({ + usable, + budget, + handlePath, + ...(options.report ? { report: options.report } : {}), + }); + if (!probe.ok) { + reporter.emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: probe.failure, + }); + return { enabled: true, failure: probe.failure }; + } + + reporter.emit(undefined, { + phase: 'started', + workloadPhase: 'applied', + containerLimitBytes: budget.containerLimitBytes, + aggregateMaxBytes: budget.aggregateMaxBytes, + reserveBytes: budget.containerLimitBytes - budget.aggregateMaxBytes, + appliedMaxBytes: budget.aggregateMaxBytes, + readbackMaxBytes: probe.placement.appliedReadbackBytes, + cpuController: probe.placement.cpuController, + siblingProtection: false, + workloadLimitSource: budget.source, + }); + return { enabled: true, placement: probe.placement }; +} + +export function applyManagedWorkloadLimits(input: { + parentReference: string; + serverReference: string; + toolsReference: string; + aggregateMaxBytes: number; +}): { cpuController: boolean } { + try { + writeFileSync(path.join(input.parentReference, 'cgroup.subtree_control'), '+memory +cpu'); + } catch { + writeFileSync(path.join(input.parentReference, 'cgroup.subtree_control'), '+memory'); + } + const subtree = readTokenSet(path.join(input.parentReference, 'cgroup.subtree_control')); + if (!subtree.ok || !subtree.tokens.has('memory')) { + throw new Error('Managed workload memory controller unavailable'); + } + + writeFileSync(path.join(input.toolsReference, 'memory.max'), String(input.aggregateMaxBytes)); + writeFileSync(path.join(input.toolsReference, 'memory.oom.group'), '1'); + writeFileSync(path.join(input.serverReference, 'memory.oom.group'), '0'); + writeControlOptional(path.join(input.toolsReference, 'memory.swap.max'), '0'); + writeControlOptional(path.join(input.toolsReference, 'cpu.weight'), String(WORKLOAD_CPU_WEIGHT)); + + const toolsMax = readMemoryMax(path.join(input.toolsReference, 'memory.max')); + if (toolsMax.kind !== 'limit' || toolsMax.bytes !== input.aggregateMaxBytes) { + throw new Error('Managed workload tool memory.max readback mismatch'); + } + const toolsOomGroup = readControl(path.join(input.toolsReference, 'memory.oom.group')); + if (!toolsOomGroup.ok || toolsOomGroup.text.trim() !== '1') { + throw new Error('Managed workload tool memory.oom.group readback mismatch'); + } + const serverOomGroup = readControl(path.join(input.serverReference, 'memory.oom.group')); + if (!serverOomGroup.ok || serverOomGroup.text.trim() !== '0') { + throw new Error('Managed workload server memory.oom.group readback mismatch'); + } + const serverMax = readControl(path.join(input.serverReference, 'memory.max')); + if (serverMax.ok ? serverMax.text.trim() !== 'max' : !serverMax.missing) { + throw new Error('Managed workload server memory.max readback mismatch'); + } + const toolsSwap = readControl(path.join(input.toolsReference, 'memory.swap.max')); + if (toolsSwap.ok ? toolsSwap.text.trim() !== '0' : !toolsSwap.missing) { + throw new Error('Managed workload tool memory.swap.max readback mismatch'); + } + return readCpuWeight(input.toolsReference); +} + +function readCpuWeight(toolsReference: string): { cpuController: boolean } { + const cpu = readControl(path.join(toolsReference, 'cpu.weight')); + return { cpuController: cpu.ok && cpu.text.trim() === String(WORKLOAD_CPU_WEIGHT) }; +} + +export function verifyWorkloadParent(placement: WorkloadPlacement): void { + const stat = fstatSync(placement.parentFd); + if (stat.dev !== placement.parentDev || stat.ino !== placement.parentIno) { + throw new WorkloadUnavailableError('readback_mismatch'); + } + const applied = readMemoryMax(path.join(placement.parentReference, 'memory.max')); + if (applied.kind !== 'limit' || applied.bytes !== placement.aggregateMaxBytes) { + throw new WorkloadUnavailableError('readback_mismatch'); + } + const oomGroup = readControl(path.join(placement.parentReference, 'memory.oom.group')); + if (!oomGroup.ok || oomGroup.text.trim() !== '0') { + throw new WorkloadUnavailableError('readback_mismatch'); + } + const swap = readControl(path.join(placement.parentReference, 'memory.swap.max')); + if (swap.ok ? swap.text.trim() !== '0' : !swap.missing) { + throw new WorkloadUnavailableError('readback_mismatch'); + } +} + +export function admitControlWorkload( + workload: ControlWorkload | undefined +): WorkloadPlacement | undefined { + if (!workload || !workload.enabled) return undefined; + if (!workload.placement) return undefined; + try { + verifyWorkloadParent(workload.placement); + } catch (error) { + createWorkloadReporter(workload.placement.report).emit(undefined, { + phase: 'failed', + workloadPhase: 'failed', + workloadFailure: error instanceof WorkloadUnavailableError ? error.failure : 'unavailable', + }); + return undefined; + } + return workload.placement; +} + +export function closeControlWorkload(workload: ControlWorkload | undefined): void { + if (!workload?.placement) return; + try { + closeSync(workload.placement.parentFd); + } catch { + return; + } +} diff --git a/services/cloud-agent-next/wrapper/src/control/worktree-runtime.ts b/services/cloud-agent-next/wrapper/src/control/worktree-runtime.ts index 67e52621dd..e8e0821f05 100644 --- a/services/cloud-agent-next/wrapper/src/control/worktree-runtime.ts +++ b/services/cloud-agent-next/wrapper/src/control/worktree-runtime.ts @@ -32,6 +32,7 @@ import { type DirectProcessObserver, type OwnedProcessScope, } from './owned-processes.js'; +import { admitControlWorkload, type ControlWorkload } from './workload-cgroup.js'; import type { NativeOperationTarget, NativeRetirement } from './session-operation-cleanup.js'; import { retireWorktreeRuntime, @@ -183,6 +184,7 @@ type ServerOptions = { env: Record; signal: AbortSignal; timeoutMs?: number; + workload?: ControlWorkload; onProcessScope?: (scope: OwnedProcessScope) => void; onProcessObserver?: (observer: DirectProcessObserver) => void; claimCleanupDeadline?: (deadlineAt?: number) => number; @@ -353,7 +355,8 @@ export async function startWorktreeKiloServer( options: ServerOptions ): Promise }> { options.signal.throwIfAborted(); - const processes = createOwnedProcessScope(); + const placement = admitControlWorkload(options.workload); + const processes = createOwnedProcessScope(placement); options.onProcessScope?.(processes); const proc = processes.spawn('kilo', ['serve', '--hostname=127.0.0.1', '--port=0'], { cwd: options.directory, @@ -558,6 +561,7 @@ function emitWorktreeRuntimeAllocation( export function createWorktreeKiloRuntimes(options: { homeRoot?: string; inheritedEnv?: NodeJS.ProcessEnv; + workload?: ControlWorkload; startServer?: (options: ServerOptions) => Promise; onEvent?: (runtime: WorktreeKiloRuntime, event: WorktreeKiloEvent) => unknown; onRootRetirementStarted?: (retirement: RootRuntimeRetirementStarted) => void; @@ -1211,6 +1215,7 @@ export function createWorktreeKiloRuntimes(options: { directory: entry.directory, env: entry.env, signal: abort.signal, + ...(options.workload ? { workload: options.workload } : {}), onProcessScope: processes => { entry.processes = processes; entry.processIssued = true;