diff --git a/services/cloud-agent-next/DEBUG.md b/services/cloud-agent-next/DEBUG.md index 50dc4b7f763..de14666b32d 100644 --- a/services/cloud-agent-next/DEBUG.md +++ b/services/cloud-agent-next/DEBUG.md @@ -180,7 +180,7 @@ logs/control////files.tar.gz For a `no_progress` or `execution_limit` outcome, find the `session.execution` record with `phase=deadline_expired`. The native session supervisor emits it for every expired Kilo execution, including unrouted and subagent work. It records `reason`, `nativeRuntimeId` and the expired `kiloSessionId`, plus `sessionId` and `rootKiloSessionId` when the execution belongs to a routed chat. `no_progress` means 20 minutes passed without real assistant or tool progress; a silent running tool does not hold that clock, and only waits on the user pause it. `execution_limit` is the separate 120-minute wall-clock cap. The wrapper no longer reads native message parts before a deadline, so there is no last-tool or `phase=freshness` probe evidence; use the Kilo CLI log in the file archive for tool detail. -`control.workload` records are per `scopeId`, so count distinct scopes to see whether the allocation was shared. `cpuUsageUsec`, `cpuThrottledUsec`, `cpuThrottleCount`, `ioReadBytes`, and `ioWriteBytes` are cumulative cgroup counters: compare consecutive samples from the same scope. `toolCpuUsageUsec`, `serverCpuUsageUsec`, `toolIoReadBytes`, and `toolIoWriteBytes` separate the tools subgroup from the Kilo server; they do not identify one command when several run together. `memoryMaxEvents` counts attempts to cross the memory limit and can rise without an OOM kill. The Worker `container_launch_identity` diagnostic joins the physical allocation reference and sandbox name to Cloudflare's 64-character `containerInstanceId`, which can be queried in Cloudflare's historical Containers metrics. These counters show resource activity or contention, not useful tool progress. +`control.workload` records are per `scopeId`, so count distinct scopes to see whether the allocation was shared. `cpuUsageUsec`, `cpuThrottledUsec`, `cpuThrottleCount`, `ioReadBytes`, and `ioWriteBytes` are cumulative cgroup counters: compare consecutive samples from the same scope. Stats sample every 30 seconds, plus sooner (at most once per 10 seconds) when a throttle or memory-event counter increases or the group first crosses 90% of its memory cap (the pressure signal releases only below 80%), so consecutive samples may be farther apart than the old one-second cadence. `toolCpuUsageUsec`, `serverCpuUsageUsec`, `toolIoReadBytes`, and `toolIoWriteBytes` separate the tools subgroup from the Kilo server; they do not identify one command when several run together. `memoryMaxEvents` counts attempts to cross the memory limit and can rise without an OOM kill. `anonBytes`, `fileBytes`, and `shmemBytes` are the group `memory.stat` split, and `toolCurrentBytes`/`serverCurrentBytes` (with peaks) attribute the total to the tools subgroup or the Kilo server, so a full headroom can be traced to page cache from tool or Docker storage I/O (`file`) versus process memory (`anon`). `pressureAvailable` reports whether the kernel exposes `memory.pressure` (PSI); when it is false, rely on `memory.events` and `memory.stat`. A `health_probe_false` restart logs the latest snapshot for that runtime's own scope under the restart line's `workload` key, and a restart with the group at 95% of its cap fails the turn as `sandbox_out_of_memory` instead of `agent_restarted`. The Worker `container_launch_identity` diagnostic joins the physical allocation reference and sandbox name to Cloudflare's 64-character `containerInstanceId`, which can be queried in Cloudflare's historical Containers metrics. These counters show resource activity or contention, not useful tool progress. List and download these JSON batches and the overwriteable `files.tar.gz` object from R2 using local tooling and the key prefix above. Each wrapper incarnation keeps one file archive; later uploads replace it. Uploaded objects remain available after the container disappears, subject to the bucket's retention policy. The legacy `getWrapperLogs` live-file reader and tarball retrieval do not read these control-plane archives. diff --git a/services/cloud-agent-next/src/control-plane/session/callbacks.ts b/services/cloud-agent-next/src/control-plane/session/callbacks.ts index 2ee610ec725..05c0dd5794d 100644 --- a/services/cloud-agent-next/src/control-plane/session/callbacks.ts +++ b/services/cloud-agent-next/src/control-plane/session/callbacks.ts @@ -66,6 +66,7 @@ const CONTROL_REASON_MESSAGES: Record = { connection_lost: 'The sandbox connection was lost', sandbox_lost: 'The sandbox was lost', agent_restarted: 'The agent restarted', + sandbox_out_of_memory: 'The sandbox ran out of memory', no_progress: 'The turn made no progress', no_outcome: 'The turn did not complete', prompt_failed: 'Prompt delivery failed', diff --git a/services/cloud-agent-next/src/shared/control-diagnostics.ts b/services/cloud-agent-next/src/shared/control-diagnostics.ts index e37ed7797b9..d5fabbf85b1 100644 --- a/services/cloud-agent-next/src/shared/control-diagnostics.ts +++ b/services/cloud-agent-next/src/shared/control-diagnostics.ts @@ -390,6 +390,13 @@ export const controlDiagnosticFieldsSchema = z readbackMaxBytes: count.optional(), currentBytes: count.optional(), peakBytes: count.optional(), + anonBytes: count.optional(), + fileBytes: count.optional(), + shmemBytes: count.optional(), + toolCurrentBytes: count.optional(), + toolPeakBytes: count.optional(), + serverCurrentBytes: count.optional(), + serverPeakBytes: count.optional(), oomKills: count.optional(), oomGroupKills: count.optional(), toolOomKills: count.optional(), @@ -399,6 +406,7 @@ export const controlDiagnosticFieldsSchema = z serverCount: count.optional(), pressureSomeTotal: count.optional(), pressureFullTotal: count.optional(), + pressureAvailable: z.boolean().optional(), memoryMaxEvents: count.optional(), memoryOomEvents: count.optional(), cpuUsageUsec: count.optional(), diff --git a/services/cloud-agent-next/src/shared/control-plane-protocol.ts b/services/cloud-agent-next/src/shared/control-plane-protocol.ts index 83644fd917c..9919988d6e0 100644 --- a/services/cloud-agent-next/src/shared/control-plane-protocol.ts +++ b/services/cloud-agent-next/src/shared/control-plane-protocol.ts @@ -106,6 +106,7 @@ export const CONTROL_PLANE_FAILURE_REASON_VALUES = [ 'connection_lost', 'sandbox_lost', 'agent_restarted', + 'sandbox_out_of_memory', 'no_progress', 'no_outcome', 'prompt_failed', diff --git a/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts b/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts index 72fb668cb62..eefa4644cbe 100644 --- a/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts +++ b/services/cloud-agent-next/src/telemetry/control-plane-failure.test.ts @@ -136,6 +136,13 @@ const cases: ReadonlyArray< ], ['sandbox_lost', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], ['agent_restarted', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_disconnected'], + [ + 'sandbox_out_of_memory', + 'accepted', + 'failed', + 'post_dispatch_no_activity', + 'wrapper_disconnected', + ], ['no_progress', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_no_output'], ['no_outcome', 'accepted', 'failed', 'post_dispatch_no_activity', 'wrapper_no_output'], [ diff --git a/services/cloud-agent-next/src/telemetry/control-plane-failure.ts b/services/cloud-agent-next/src/telemetry/control-plane-failure.ts index 1cbf99f158c..ed14a7a2517 100644 --- a/services/cloud-agent-next/src/telemetry/control-plane-failure.ts +++ b/services/cloud-agent-next/src/telemetry/control-plane-failure.ts @@ -106,9 +106,12 @@ export function classifyControlPlaneFailure( return dispatchState === 'accepted' ? POST_DISPATCH_WRAPPER_DISCONNECTED : PRE_DISPATCH_KILO_SERVER; + // `sandbox_out_of_memory` is a health restart with the workload at its cap: + // still a post-dispatch wrapper-path failure, but with a reason triage can name. case 'connection_lost': case 'sandbox_lost': case 'agent_restarted': + case 'sandbox_out_of_memory': return POST_DISPATCH_WRAPPER_DISCONNECTED; case 'no_progress': case 'no_outcome': diff --git a/services/cloud-agent-next/wrapper/src/control-plane/kilo-health-diagnostic.test.ts b/services/cloud-agent-next/wrapper/src/control-plane/kilo-health-diagnostic.test.ts index 3ec6eeac403..bed3da81fdf 100644 --- a/services/cloud-agent-next/wrapper/src/control-plane/kilo-health-diagnostic.test.ts +++ b/services/cloud-agent-next/wrapper/src/control-plane/kilo-health-diagnostic.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, spyOn } from 'bun:test'; import { CONTROL_PLANE_TIMERS } from '../../../src/shared/control-plane-timers.js'; +import type { WorkloadSnapshot } from '../control/workload-cgroup.js'; import { createKiloRuntime, type KiloRuntimeScheduler } from './kilo-runtime.js'; const SENTINEL = 'PRIVATE_HEALTH_SENTINEL'; @@ -86,6 +87,7 @@ function harness( healthRequestMs?: number; failReplacement?: boolean; isIdle?: () => Promise; + latestWorkloadSnapshot?: () => WorkloadSnapshot | undefined; } = {} ) { let clock = 0; @@ -101,6 +103,7 @@ function harness( }, }; const counts = { spawns: 0, stops: 0, opens: 0, restarts: 0 }; + const restartInfos: Array<{ reason: string; outcomeReason?: string }> = []; const logs: string[] = []; const exits: Array<() => void> = []; let unavailable = 0; @@ -117,6 +120,9 @@ function harness( prepareFilesystem: async () => {}, readProcessStartTime: () => undefined, ...(options.isIdle ? { isIdle: options.isIdle } : {}), + ...(options.latestWorkloadSnapshot + ? { latestWorkloadSnapshot: options.latestWorkloadSnapshot } + : {}), spawnKilo: async () => { counts.spawns++; if (options.failReplacement && counts.spawns > 1) throw new Error('replacement failed'); @@ -144,8 +150,12 @@ function harness( }; }, log: line => logs.push(line), - onRestart: () => { + onRestart: info => { counts.restarts++; + restartInfos.push({ + reason: info.reason, + ...(info.outcomeReason === undefined ? {} : { outcomeReason: info.outcomeReason }), + }); }, onUnavailable: () => { unavailable++; @@ -156,6 +166,7 @@ function harness( logs, counts, exits, + restartInfos, unavailable: () => unavailable, silence(ms = 30000) { clock += ms; @@ -599,4 +610,64 @@ describe('default SDK health diagnostics', () => { await test.runtime.shutdown(); } }); + + it('attaches the latest workload snapshot and names memory exhaustion on a health restart', async () => { + const gib = 1024 ** 3; + const snapshot: WorkloadSnapshot = { + aggregateMaxBytes: 11 * gib, + toolsMaxBytes: 8 * gib, + containerLimitBytes: 12 * gib, + currentBytes: 11 * gib, + anonBytes: 9 * gib, + fileBytes: 2 * gib, + shmemBytes: 0, + toolCurrentBytes: 4 * gib, + serverCurrentBytes: 7 * gib, + pressureAvailable: true, + oomKills: 0, + oomGroupKills: 0, + toolOomKills: 0, + serverOomKills: 0, + }; + const server = Bun.serve({ + hostname: '127.0.0.1', + port: 0, + fetch() { + return Response.json({ healthy: false, private: SENTINEL }); + }, + }); + const test = harness(String(server.url), { latestWorkloadSnapshot: () => snapshot }); + try { + await test.runtime.ensure(); + test.silence(); + await waitFor(() => test.counts.restarts === 1); + const line = test.logs.find(candidate => candidate.includes(' diagnostic=')); + expect(line).toBeDefined(); + const marker = ' diagnostic='; + const serialized = line!.slice(line!.lastIndexOf(marker) + marker.length); + expect(Buffer.byteLength(serialized)).toBeLessThanOrEqual(512); + expect(serialized).not.toContain(SENTINEL); + const diagnostic = JSON.parse(serialized) as { + trigger: string; + workload?: Record; + }; + expect(diagnostic.trigger).toBe('health_probe_false'); + expect(diagnostic.workload).toEqual({ + currentBytes: 11 * gib, + aggregateMaxBytes: 11 * gib, + anonBytes: 9 * gib, + fileBytes: 2 * gib, + shmemBytes: 0, + toolCurrentBytes: 4 * gib, + serverCurrentBytes: 7 * gib, + pressureAvailable: true, + }); + expect(test.restartInfos).toEqual([ + { reason: 'hang', outcomeReason: 'sandbox_out_of_memory' }, + ]); + } finally { + await test.runtime.shutdown(); + await server.stop(true); + } + }); }); diff --git a/services/cloud-agent-next/wrapper/src/control-plane/kilo-runtime.ts b/services/cloud-agent-next/wrapper/src/control-plane/kilo-runtime.ts index 8ba4c215426..861a28cf58d 100644 --- a/services/cloud-agent-next/wrapper/src/control-plane/kilo-runtime.ts +++ b/services/cloud-agent-next/wrapper/src/control-plane/kilo-runtime.ts @@ -13,7 +13,12 @@ import type { ControlPlaneTimers } from '../../../src/shared/control-plane-timer import { createWrapperKiloClient, type WrapperKiloClient } from '../kilo-api.js'; import { createOwnedProcessScope, type OwnedProcessScope } from '../control/owned-processes.js'; import { isKiloServerProcess } from '../tool-cgroup.js'; -import { admitControlWorkload, type ControlWorkload } from '../control/workload-cgroup.js'; +import { + admitControlWorkload, + isWorkloadAtCap, + type ControlWorkload, + type WorkloadSnapshot, +} from '../control/workload-cgroup.js'; import { createKiloEventFeed, type KiloEventFeed, @@ -154,6 +159,7 @@ export type KiloRestartInfo = { directory: string; reason: KiloRestartReason; interruptedExecutions?: ExecutionIdentity[]; + outcomeReason?: string; }; /** @@ -263,6 +269,11 @@ export type KiloRuntimeOptions = { readSnapshot?: typeof readSessionSnapshot; /** Fired after Kilo comes back with a fresh process; B8 hands over busy turns. */ onRestart?: (info: KiloRestartInfo) => void; + /** + * Overrides the workload snapshot source. Production reads the runtime's own + * process scope; tests that cannot spawn one inject a snapshot here. + */ + latestWorkloadSnapshot?: () => WorkloadSnapshot | undefined; /** Fired once when the 3-in-10-minutes budget is spent (spec ยง7 "Kilo supervision"). */ onUnavailable?: (directory: string) => void; scheduler?: KiloRuntimeScheduler; @@ -478,6 +489,11 @@ export function createKiloRuntime(options: KiloRuntimeOptions): KiloRuntime { })); let kiloProcess: KiloProcess | undefined; + let latestScope: OwnedProcessScope | undefined; + // This runtime's own scope, never a sibling's: a restart must be attributed to + // the memory this runtime's Kilo and tools actually used. + const latestWorkloadSnapshot = (): WorkloadSnapshot | undefined => + latestScope?.latestSnapshot() ?? options.latestWorkloadSnapshot?.(); let pidfilePath: string | undefined; let client: WrapperKiloClient | undefined; let feed: KiloEventFeed | undefined; @@ -760,6 +776,9 @@ export function createKiloRuntime(options: KiloRuntimeOptions): KiloRuntime { env, signal: controller.signal, ...(options.workload ? { workload: options.workload } : {}), + onProcessScope: scope => { + latestScope = scope; + }, }); } catch (error) { clearTimeout(deadline); @@ -884,6 +903,11 @@ export function createKiloRuntime(options: KiloRuntimeOptions): KiloRuntime { const interruptedExecutions = activity?.executions().filter(execution => execution.activity !== 'stopping') ?? []; phase = 'restarting'; + const restartWorkload = trigger === 'health_probe_false' ? latestWorkloadSnapshot() : undefined; + const outcomeReason = + restartWorkload !== undefined && isWorkloadAtCap(restartWorkload) + ? ('sandbox_out_of_memory' as const) + : undefined; let diagnostic = ''; if (reason !== 'credentials') { const observation = trigger === 'health_probe_false' ? healthObservation : undefined; @@ -905,6 +929,20 @@ export function createKiloRuntime(options: KiloRuntimeOptions): KiloRuntime { sseSilenceMs: bound(silence), sseSilenceClamped: bound(silence) !== silence, reconnectCount: Math.min(6, reconnects.length), + ...(trigger === 'health_probe_false' && restartWorkload !== undefined + ? { + workload: { + currentBytes: restartWorkload.currentBytes, + aggregateMaxBytes: restartWorkload.aggregateMaxBytes, + anonBytes: restartWorkload.anonBytes, + fileBytes: restartWorkload.fileBytes, + shmemBytes: restartWorkload.shmemBytes, + toolCurrentBytes: restartWorkload.toolCurrentBytes, + serverCurrentBytes: restartWorkload.serverCurrentBytes, + pressureAvailable: restartWorkload.pressureAvailable, + }, + } + : {}), })}`; } log( @@ -961,7 +999,12 @@ export function createKiloRuntime(options: KiloRuntimeOptions): KiloRuntime { }); } try { - options.onRestart?.({ directory: options.directory, reason, interruptedExecutions }); + options.onRestart?.({ + directory: options.directory, + reason, + interruptedExecutions, + ...(outcomeReason === undefined ? {} : { outcomeReason }), + }); } catch (error) { log( `control-plane kilo restart handler failed directory=${options.directory} error=${ diff --git a/services/cloud-agent-next/wrapper/src/control-plane/turn.test.ts b/services/cloud-agent-next/wrapper/src/control-plane/turn.test.ts index ac25b7939f8..5cda25c34af 100644 --- a/services/cloud-agent-next/wrapper/src/control-plane/turn.test.ts +++ b/services/cloud-agent-next/wrapper/src/control-plane/turn.test.ts @@ -934,6 +934,29 @@ describe('turn resubmission', () => { expect(outcomeFrames(h.frames)[0]).toMatchObject({ reason: 'agent_restarted' }); }); + it('reports the runtime-supplied memory reason instead of agent_restarted', async () => { + const h = createHarness(); + h.registerRoute(routeSpec()); + const client = h.client(routeSpec()); + h.manager.submit(SESSION_ID, promptPayload('m1')); + await settle(); + h.setState(KILO_SESSION, runningState); + h.manager.observeKiloEvent( + kiloEvent('message.part.updated', { + part: { sessionID: KILO_SESSION, messageID: 'assistant-1', type: 'tool', tool: 'bash' }, + }) + ); + h.manager.onRuntimeRestart({ + directory: DIRECTORY, + reason: 'hang', + key: DIRECTORY, + outcomeReason: 'sandbox_out_of_memory', + }); + await settle(); + expect(client.prompts).toHaveLength(1); + expect(outcomeFrames(h.frames)[0]).toMatchObject({ reason: 'sandbox_out_of_memory' }); + }); + it('resubmits only the restarted per-session runtime, not its sibling', async () => { const h = createHarness(); const specA = routeSpec({ diff --git a/services/cloud-agent-next/wrapper/src/control-plane/turn.ts b/services/cloud-agent-next/wrapper/src/control-plane/turn.ts index e851e8525a8..4eb74b6a1a9 100644 --- a/services/cloud-agent-next/wrapper/src/control-plane/turn.ts +++ b/services/cloud-agent-next/wrapper/src/control-plane/turn.ts @@ -997,6 +997,7 @@ export function createTurnManager(deps: TurnManagerDeps) { onRuntimeRestart(info: KiloRestartInfo & { key: string }): void { const ownedTurns = new Set(turnsForRuntimeKey(info.key).map(turn => turn.route.sessionId)); void publishCommandsForRuntimeKey(info.key); + const outcomeReason = info.outcomeReason ?? 'agent_restarted'; for (const turn of turnsForRuntimeKey(info.key)) { turn.deferredCompletion = undefined; if (turn.phase === 'finalizing') { @@ -1005,7 +1006,7 @@ export function createTurnManager(deps: TurnManagerDeps) { // started; the restart interrupted its Kilo work, so fail it // instead of letting it end as no_progress. turn.stepAbort?.abort(new Error('agent restarted')); - sendOutcome(turn, 'failed', 'agent_restarted'); + sendOutcome(turn, 'failed', outcomeReason); } // Otherwise let finalization finish; a later idle triggers it again. continue; @@ -1036,7 +1037,7 @@ export function createTurnManager(deps: TurnManagerDeps) { drainInbox(turn); continue; } - sendOutcome(turn, 'failed', 'agent_restarted'); + sendOutcome(turn, 'failed', outcomeReason); } const reported = new Set(); for (const execution of info.interruptedExecutions ?? []) { 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 8448c618fb4..83d7e364281 100644 --- a/services/cloud-agent-next/wrapper/src/control/owned-processes.ts +++ b/services/cloud-agent-next/wrapper/src/control/owned-processes.ts @@ -43,6 +43,7 @@ import { CGROUP_FS_MAGIC, classifyWorkloadMembers, createWorkloadReporter, + decideWorkloadStatsEmission, KILO_OOM_SCORE_ADJ, readWorkloadStats, TOOL_OOM_SCORE_ADJ, @@ -52,6 +53,8 @@ import { writeOomScoreAdj, type WorkloadPlacement, type WorkloadProcessEntry, + type WorkloadSnapshot, + type WorkloadStatsEmissionState, } from './workload-cgroup.js'; export type DirectProcessState = 'absent' | 'reused' | 'alive' | 'unknown'; @@ -75,6 +78,7 @@ export type OwnedProcessScope = { captureBaseline(allowed: (argv: string[]) => boolean, deadlineAt?: number): Promise; verify(baseline?: boolean, deadlineAt?: number): Promise; stop(deadlineAt: number): Promise; + latestSnapshot(): WorkloadSnapshot | undefined; }; type ProcessIdentity = { @@ -643,6 +647,11 @@ export function createOwnedProcessScope(placement?: WorkloadPlacement): OwnedPro let sweeping = false; let lastOomKills = 0; let lastOomGroupKills = 0; + let statsEmission: WorkloadStatsEmissionState = { + lastEmittedAtMs: 0, + highPressure: false, + }; + let workloadSnapshot: WorkloadSnapshot | undefined; const children = new Set(); const baseline = new Set(); const observations = new Set(); @@ -738,6 +747,40 @@ export function createOwnedProcessScope(placement?: WorkloadPlacement): OwnedPro const stats = readWorkloadStats(group.reference); const toolStats = readWorkloadStats(managed.toolsReference); const serverStats = readWorkloadStats(managed.serverReference); + if (placement !== undefined) { + workloadSnapshot = { + scopeId, + aggregateMaxBytes: placement.aggregateMaxBytes, + toolsMaxBytes: placement.toolsMaxBytes, + containerLimitBytes: placement.containerLimitBytes, + oomKills: stats.oomKills, + oomGroupKills: stats.oomGroupKills, + toolOomKills: toolStats.oomKills, + serverOomKills: serverStats.oomKills, + pressureAvailable: stats.pressureAvailable, + ...(stats.currentBytes !== undefined ? { currentBytes: stats.currentBytes } : {}), + ...(stats.peakBytes !== undefined ? { peakBytes: stats.peakBytes } : {}), + ...(stats.anonBytes !== undefined ? { anonBytes: stats.anonBytes } : {}), + ...(stats.fileBytes !== undefined ? { fileBytes: stats.fileBytes } : {}), + ...(stats.shmemBytes !== undefined ? { shmemBytes: stats.shmemBytes } : {}), + ...(toolStats.currentBytes !== undefined + ? { toolCurrentBytes: toolStats.currentBytes } + : {}), + ...(toolStats.peakBytes !== undefined ? { toolPeakBytes: toolStats.peakBytes } : {}), + ...(serverStats.currentBytes !== undefined + ? { serverCurrentBytes: serverStats.currentBytes } + : {}), + ...(serverStats.peakBytes !== undefined + ? { serverPeakBytes: serverStats.peakBytes } + : {}), + ...(stats.memoryMaxEvents !== undefined + ? { memoryMaxEvents: stats.memoryMaxEvents } + : {}), + ...(stats.memoryOomEvents !== undefined + ? { memoryOomEvents: stats.memoryOomEvents } + : {}), + }; + } if (stats.oomKills > lastOomKills || stats.oomGroupKills > lastOomGroupKills) { lastOomKills = Math.max(lastOomKills, stats.oomKills); lastOomGroupKills = Math.max(lastOomGroupKills, stats.oomGroupKills); @@ -750,6 +793,14 @@ export function createOwnedProcessScope(placement?: WorkloadPlacement): OwnedPro serverOomKills: serverStats.oomKills, }); } + const decision = decideWorkloadStatsEmission({ + nowMs: Date.now(), + stats, + limitBytes: activePlacement?.aggregateMaxBytes, + state: statsEmission, + }); + statsEmission = decision.state; + if (!decision.emit) return; workloadReporter?.emit(scopeId, { phase: 'completed', workloadPhase: 'stats', @@ -763,6 +814,10 @@ export function createOwnedProcessScope(placement?: WorkloadPlacement): OwnedPro cpuController: managed.cpuController, ...(stats.currentBytes !== undefined ? { currentBytes: stats.currentBytes } : {}), ...(stats.peakBytes !== undefined ? { peakBytes: stats.peakBytes } : {}), + ...(stats.anonBytes !== undefined ? { anonBytes: stats.anonBytes } : {}), + ...(stats.fileBytes !== undefined ? { fileBytes: stats.fileBytes } : {}), + ...(stats.shmemBytes !== undefined ? { shmemBytes: stats.shmemBytes } : {}), + pressureAvailable: stats.pressureAvailable, ...(stats.pressureSomeTotal !== undefined ? { pressureSomeTotal: stats.pressureSomeTotal } : {}), @@ -790,6 +845,14 @@ export function createOwnedProcessScope(placement?: WorkloadPlacement): OwnedPro ...(toolStats.ioWriteBytes !== undefined ? { toolIoWriteBytes: toolStats.ioWriteBytes } : {}), + ...(toolStats.currentBytes !== undefined + ? { toolCurrentBytes: toolStats.currentBytes } + : {}), + ...(toolStats.peakBytes !== undefined ? { toolPeakBytes: toolStats.peakBytes } : {}), + ...(serverStats.currentBytes !== undefined + ? { serverCurrentBytes: serverStats.currentBytes } + : {}), + ...(serverStats.peakBytes !== undefined ? { serverPeakBytes: serverStats.peakBytes } : {}), }); } catch { workloadReporter?.emit(path.basename(group.directory), { @@ -1051,6 +1114,7 @@ export function createOwnedProcessScope(placement?: WorkloadPlacement): OwnedPro }, run: operation => current.run(scope, operation), observesOccupancy: occupancyObservable, + latestSnapshot: () => workloadSnapshot, seal() { sealed = true; }, diff --git a/services/cloud-agent-next/wrapper/src/control/workload-cgroup-stats.test.ts b/services/cloud-agent-next/wrapper/src/control/workload-cgroup-stats.test.ts index 2d6c5364669..41913e0bc1b 100644 --- a/services/cloud-agent-next/wrapper/src/control/workload-cgroup-stats.test.ts +++ b/services/cloud-agent-next/wrapper/src/control/workload-cgroup-stats.test.ts @@ -2,7 +2,19 @@ import { afterEach, describe, expect, it } from 'bun:test'; import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import path from 'node:path'; -import { readWorkloadStats } from './workload-cgroup.js'; +import { + decideWorkloadStatsEmission, + isWorkloadAtCap, + readWorkloadStats, + WORKLOAD_AT_CAP_FRACTION, + WORKLOAD_MEMORY_PRESSURE_FRACTION, + WORKLOAD_MEMORY_PRESSURE_RELEASE_FRACTION, + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + WORKLOAD_STATS_INTERVAL_MS, + type WorkloadSnapshot, + type WorkloadStats, + type WorkloadStatsEmissionState, +} from './workload-cgroup.js'; const directories: string[] = []; @@ -21,6 +33,10 @@ describe('readWorkloadStats', () => { path.join(directory, 'memory.events'), 'max 4\noom 2\noom_kill 1\noom_group_kill 0\n' ); + writeFileSync( + path.join(directory, 'memory.stat'), + 'anon 700\nfile 300\nshmem 50\ninactive_file 200\n' + ); writeFileSync( path.join(directory, 'memory.pressure'), 'some avg10=0 total=50\nfull avg10=0 total=20\n' @@ -37,10 +53,14 @@ describe('readWorkloadStats', () => { expect(readWorkloadStats(directory)).toEqual({ currentBytes: 1024, peakBytes: 2048, + anonBytes: 700, + fileBytes: 300, + shmemBytes: 50, memoryMaxEvents: 4, memoryOomEvents: 2, oomKills: 1, oomGroupKills: 0, + pressureAvailable: true, pressureSomeTotal: 50, pressureFullTotal: 20, cpuUsageUsec: 900, @@ -50,4 +70,187 @@ describe('readWorkloadStats', () => { ioWriteBytes: 240, }); }); + + it('reports absent PSI as unavailable and omits the split without memory.stat', () => { + const directory = mkdtempSync(path.join(tmpdir(), 'workload-stats-')); + directories.push(directory); + writeFileSync(path.join(directory, 'memory.current'), '512\n'); + + expect(readWorkloadStats(directory)).toEqual({ + currentBytes: 512, + oomKills: 0, + oomGroupKills: 0, + pressureAvailable: false, + }); + }); +}); + +describe('decideWorkloadStatsEmission', () => { + const baseStats: WorkloadStats = { oomKills: 0, oomGroupKills: 0, pressureAvailable: false }; + const emitted: WorkloadStatsEmissionState = { + lastEmittedAtMs: 1_000_000, + lastThrottleCount: 0, + lastMemoryMaxEvents: 0, + lastMemoryOomEvents: 0, + highPressure: false, + }; + + it('emits an initial sample and then holds the periodic cadence', () => { + const first = decideWorkloadStatsEmission({ + nowMs: 1_000_000, + stats: baseStats, + state: { lastEmittedAtMs: 0, highPressure: false }, + }); + expect(first).toMatchObject({ emit: true, reason: 'interval' }); + + const quiet = decideWorkloadStatsEmission({ + nowMs: 1_000_000 + WORKLOAD_STATS_INTERVAL_MS - 1, + stats: baseStats, + state: first.state, + }); + expect(quiet).toMatchObject({ emit: false, reason: 'none' }); + + const due = decideWorkloadStatsEmission({ + nowMs: 1_000_000 + WORKLOAD_STATS_INTERVAL_MS, + stats: baseStats, + state: quiet.state, + }); + expect(due).toMatchObject({ emit: true, reason: 'interval' }); + }); + + it('emits as soon as a throttle counter increases', () => { + const decision = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, cpuThrottleCount: 4 }, + state: emitted, + }); + expect(decision).toMatchObject({ emit: true, reason: 'throttle_increase' }); + }); + + it('holds immediate triggers to the cooldown under sustained pressure', () => { + const first = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, cpuThrottleCount: 4 }, + state: emitted, + }); + expect(first).toMatchObject({ emit: true, reason: 'throttle_increase' }); + + const tooSoon = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS + 1000, + stats: { ...baseStats, cpuThrottleCount: 9 }, + state: first.state, + }); + expect(tooSoon).toMatchObject({ emit: false, reason: 'none' }); + + const later = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + 2 * WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, cpuThrottleCount: 9 }, + state: tooSoon.state, + }); + expect(later).toMatchObject({ emit: true, reason: 'throttle_increase' }); + }); + + it('emits as soon as a memory-event counter increases', () => { + const decision = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, memoryMaxEvents: 2 }, + state: emitted, + }); + expect(decision).toMatchObject({ emit: true, reason: 'memory_events' }); + }); + + it('emits once when the group first crosses the memory-pressure threshold', () => { + const limitBytes = 1_000_000; + const pressured = Math.ceil(limitBytes * WORKLOAD_MEMORY_PRESSURE_FRACTION); + const crossed = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, currentBytes: pressured }, + limitBytes, + state: emitted, + }); + expect(crossed).toMatchObject({ emit: true, reason: 'memory_pressure' }); + expect(crossed.state.highPressure).toBe(true); + + const held = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + 2 * WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, currentBytes: pressured + 1000 }, + limitBytes, + state: crossed.state, + }); + expect(held).toMatchObject({ emit: false, reason: 'none' }); + expect(held.state.highPressure).toBe(true); + }); + + it('releases pressure only below the release fraction before it can re-arm', () => { + const limitBytes = 1_000_000; + const pressured = Math.ceil(limitBytes * WORKLOAD_MEMORY_PRESSURE_FRACTION); + const released = Math.floor(limitBytes * WORKLOAD_MEMORY_PRESSURE_RELEASE_FRACTION) - 1; + const latched: WorkloadStatsEmissionState = { ...emitted, highPressure: true }; + + const below = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, currentBytes: released }, + limitBytes, + state: latched, + }); + expect(below).toMatchObject({ emit: false, reason: 'none' }); + expect(below.state.highPressure).toBe(false); + + const rearmed = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, currentBytes: pressured }, + limitBytes, + state: below.state, + }); + expect(rearmed).toMatchObject({ emit: true, reason: 'memory_pressure' }); + }); + + it('does not treat elevated memory as pressure without a known limit', () => { + const decision = decideWorkloadStatsEmission({ + nowMs: emitted.lastEmittedAtMs + WORKLOAD_STATS_EVENT_COOLDOWN_MS, + stats: { ...baseStats, currentBytes: Number.MAX_SAFE_INTEGER }, + state: emitted, + }); + expect(decision).toMatchObject({ emit: false, reason: 'none' }); + }); +}); + +describe('isWorkloadAtCap', () => { + const gib = 1024 ** 3; + const snapshot: WorkloadSnapshot = { + aggregateMaxBytes: 11 * gib, + toolsMaxBytes: 8 * gib, + containerLimitBytes: 12 * gib, + currentBytes: 11 * gib, + pressureAvailable: false, + oomKills: 0, + oomGroupKills: 0, + toolOomKills: 0, + serverOomKills: 0, + }; + + it('treats the group at its cap as exhausted', () => { + expect(isWorkloadAtCap(snapshot)).toBe(true); + }); + + it('treats the cap threshold as exhausted and steps just below it as not', () => { + const max = 1000; + const atThreshold: WorkloadSnapshot = { + ...snapshot, + aggregateMaxBytes: max, + currentBytes: Math.ceil(max * WORKLOAD_AT_CAP_FRACTION), + }; + expect(isWorkloadAtCap(atThreshold)).toBe(true); + expect( + isWorkloadAtCap({ + ...atThreshold, + currentBytes: Math.floor(max * WORKLOAD_AT_CAP_FRACTION) - 1, + }) + ).toBe(false); + }); + + it('is false below the cap or without a reading', () => { + expect(isWorkloadAtCap({ ...snapshot, currentBytes: 5 * gib })).toBe(false); + expect(isWorkloadAtCap({ ...snapshot, currentBytes: undefined })).toBe(false); + }); }); diff --git a/services/cloud-agent-next/wrapper/src/control/workload-cgroup.ts b/services/cloud-agent-next/wrapper/src/control/workload-cgroup.ts index 0eb0f4aabc4..768234c9cce 100644 --- a/services/cloud-agent-next/wrapper/src/control/workload-cgroup.ts +++ b/services/cloud-agent-next/wrapper/src/control/workload-cgroup.ts @@ -31,6 +31,26 @@ export const KILO_SERVER_HEADROOM_BYTES = 1536 * 1024 * 1024; export const MIN_WORKLOAD_CAP_BYTES = 1024 * 1024 * 1024; export const WORKLOAD_CPU_WEIGHT = 50; export const WORKLOAD_SWEEP_INTERVAL_MS = 1000; +/** + * Stats are observation only, so they sample far slower than the containment + * sweep. The sweep still runs every second to migrate new processes; emitting + * its monotonic cpu/io/memory counters at that rate floods the diagnostic stream. + */ +export const WORKLOAD_STATS_INTERVAL_MS = 30_000; +/** + * Bounds the immediate triggers (throttle/memory-event increases, first + * pressure crossing). Those counters are monotonic and can increase on + * consecutive one-second sweeps, so without a cooldown a scope parked at its + * limit would emit on almost every sweep. + */ +export const WORKLOAD_STATS_EVENT_COOLDOWN_MS = 10_000; +export const WORKLOAD_MEMORY_PRESSURE_FRACTION = 0.9; +/** + * Pressure latches at `WORKLOAD_MEMORY_PRESSURE_FRACTION` and only releases + * below this fraction, so `memory.current` oscillating around the threshold + * cannot re-arm the trigger every sweep. + */ +export const WORKLOAD_MEMORY_PRESSURE_RELEASE_FRACTION = 0.8; export const WORKLOAD_PARENT_NAME = 'kilo-workloads'; export const WORKLOAD_SERVER_NAME = 'server'; export const WORKLOAD_TOOLS_NAME = 'tools'; @@ -92,10 +112,15 @@ export type WorkloadProcessEntry = { pid: number; ppid: number; argv: string[] } export type WorkloadStats = { currentBytes?: number; peakBytes?: number; + anonBytes?: number; + fileBytes?: number; + shmemBytes?: number; memoryMaxEvents?: number; memoryOomEvents?: number; oomKills: number; oomGroupKills: number; + /** Whether the kernel exposes cgroup v2 `memory.pressure` (PSI). */ + pressureAvailable: boolean; pressureSomeTotal?: number; pressureFullTotal?: number; cpuUsageUsec?: number; @@ -105,6 +130,135 @@ export type WorkloadStats = { ioWriteBytes?: number; }; +/** + * The last observed workload state for one scope. Unlike an emitted stats + * record it is updated on every sweep, so a restart decision made between + * samples still sees the current attribution. + */ +export type WorkloadSnapshot = { + scopeId?: string; + aggregateMaxBytes: number; + toolsMaxBytes: number; + containerLimitBytes: number; + currentBytes?: number; + peakBytes?: number; + anonBytes?: number; + fileBytes?: number; + shmemBytes?: number; + toolCurrentBytes?: number; + toolPeakBytes?: number; + serverCurrentBytes?: number; + serverPeakBytes?: number; + pressureAvailable: boolean; + memoryMaxEvents?: number; + memoryOomEvents?: number; + oomKills: number; + oomGroupKills: number; + toolOomKills: number; + serverOomKills: number; +}; + +/** + * A workload at this share of its cap is treated as memory-exhausted: the group + * cannot grow further without reclaim, so a restart at this point is attributed + * to the user's workload rather than an unexplained platform fault. + */ +export const WORKLOAD_AT_CAP_FRACTION = 0.95; + +export function isWorkloadAtCap(snapshot: WorkloadSnapshot): boolean { + const current = snapshot.currentBytes; + if (current === undefined) return false; + return current >= snapshot.aggregateMaxBytes * WORKLOAD_AT_CAP_FRACTION; +} + +export type WorkloadStatsEmissionState = { + lastEmittedAtMs: number; + lastThrottleCount?: number; + lastMemoryMaxEvents?: number; + lastMemoryOomEvents?: number; + highPressure: boolean; +}; + +export type WorkloadStatsEmissionReason = + | 'interval' + | 'throttle_increase' + | 'memory_events' + | 'memory_pressure' + | 'none'; + +export type WorkloadStatsEmissionDecision = { + emit: boolean; + reason: WorkloadStatsEmissionReason; + state: WorkloadStatsEmissionState; +}; + +/** + * Periodic stats are capped to `WORKLOAD_STATS_INTERVAL_MS`, but an actionable + * change (a new throttle or memory-event counter, or the first crossing into + * high memory pressure) emits sooner so the signal is not delayed to the next + * sample. Immediate emissions share `WORKLOAD_STATS_EVENT_COOLDOWN_MS` because + * the underlying counters are monotonic and would otherwise fire on every + * one-second sweep. Emission baselines advance only when a record is emitted, + * so an increase observed between samples still surfaces on the next decision. + */ +export function decideWorkloadStatsEmission(input: { + nowMs: number; + stats: WorkloadStats; + limitBytes?: number; + state: WorkloadStatsEmissionState; +}): WorkloadStatsEmissionDecision { + const { nowMs, stats, limitBytes, state } = input; + const current = stats.currentBytes; + const abovePressure = + limitBytes !== undefined && + current !== undefined && + current >= limitBytes * WORKLOAD_MEMORY_PRESSURE_FRACTION; + const highPressure = + abovePressure || + (state.highPressure && + limitBytes !== undefined && + current !== undefined && + current >= limitBytes * WORKLOAD_MEMORY_PRESSURE_RELEASE_FRACTION); + const throttleIncreased = + stats.cpuThrottleCount !== undefined && + state.lastThrottleCount !== undefined && + stats.cpuThrottleCount > state.lastThrottleCount; + const memoryEventsIncreased = + (stats.memoryMaxEvents !== undefined && + state.lastMemoryMaxEvents !== undefined && + stats.memoryMaxEvents > state.lastMemoryMaxEvents) || + (stats.memoryOomEvents !== undefined && + state.lastMemoryOomEvents !== undefined && + stats.memoryOomEvents > state.lastMemoryOomEvents); + const sinceEmittedMs = nowMs - state.lastEmittedAtMs; + const reason: WorkloadStatsEmissionReason = + sinceEmittedMs >= WORKLOAD_STATS_INTERVAL_MS + ? 'interval' + : sinceEmittedMs < WORKLOAD_STATS_EVENT_COOLDOWN_MS + ? 'none' + : throttleIncreased + ? 'throttle_increase' + : memoryEventsIncreased + ? 'memory_events' + : abovePressure && !state.highPressure + ? 'memory_pressure' + : 'none'; + if (reason === 'none') { + return { emit: false, reason, state: { ...state, highPressure } }; + } + return { + emit: true, + reason, + state: { + lastEmittedAtMs: nowMs, + lastThrottleCount: stats.cpuThrottleCount, + lastMemoryMaxEvents: stats.memoryMaxEvents, + lastMemoryOomEvents: stats.memoryOomEvents, + highPressure, + }, + }; +} + export class WorkloadUnavailableError extends Error { constructor(readonly failure: WorkloadFailure) { super(`Control workload unavailable: ${failure}`); @@ -152,19 +306,27 @@ export function createWorkloadReporter(report?: ControlDiagnosticReporter): Work fields.serverOomKills ?? '', fields.currentBytes ?? '', fields.peakBytes ?? '', - fields.pressureSomeTotal ?? '', - fields.pressureFullTotal ?? '', + fields.anonBytes ?? '', + fields.fileBytes ?? '', + fields.shmemBytes ?? '', fields.memoryMaxEvents ?? '', fields.memoryOomEvents ?? '', + fields.pressureSomeTotal ?? '', + fields.pressureFullTotal ?? '', fields.cpuUsageUsec ?? '', fields.cpuThrottledUsec ?? '', fields.cpuThrottleCount ?? '', fields.ioReadBytes ?? '', fields.ioWriteBytes ?? '', + fields.toolCurrentBytes ?? '', + fields.toolPeakBytes ?? '', + fields.serverCurrentBytes ?? '', + fields.serverPeakBytes ?? '', fields.toolCpuUsageUsec ?? '', fields.serverCpuUsageUsec ?? '', fields.toolIoReadBytes ?? '', fields.toolIoWriteBytes ?? '', + fields.pressureAvailable ?? '', fields.toolCount ?? '', fields.serverCount ?? '', fields.migratedCount ?? '', @@ -282,11 +444,12 @@ export function parsePressureTotal( } export function readWorkloadStats(reference: string): WorkloadStats { - const stats: WorkloadStats = { oomKills: 0, oomGroupKills: 0 }; + const stats: WorkloadStats = { oomKills: 0, oomGroupKills: 0, pressureAvailable: false }; 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')); + const memoryStat = readControl(path.join(reference, 'memory.stat')); const cpu = readControl(path.join(reference, 'cpu.stat')); const io = readControl(path.join(reference, 'io.stat')); if (current.ok) { @@ -297,10 +460,21 @@ export function readWorkloadStats(reference: string): WorkloadStats { const parsed = Number.parseInt(peak.text.trim(), 10); if (Number.isSafeInteger(parsed) && parsed >= 0) stats.peakBytes = parsed; } + stats.pressureAvailable = pressure.ok; 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 (memoryStat.ok) { + for (const line of memoryStat.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 === 'anon') stats.anonBytes = parsed; + if (key === 'file') stats.fileBytes = parsed; + if (key === 'shmem') stats.shmemBytes = parsed; + } + } if (events.ok) { for (const line of events.text.split('\n')) { const [key, value] = line.trim().split(/\s+/);