Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion services/cloud-agent-next/DEBUG.md
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,7 @@ logs/control/<sandboxId>/<allocationId>/<wrapperInstanceId>/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.

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ const CONTROL_REASON_MESSAGES: Record<ControlPlaneFailureReason, string> = {
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',
Expand Down
8 changes: 8 additions & 0 deletions services/cloud-agent-next/src/shared/control-diagnostics.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand All @@ -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(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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'],
[
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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':
Expand Down
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -86,6 +87,7 @@ function harness(
healthRequestMs?: number;
failReplacement?: boolean;
isIdle?: () => Promise<boolean>;
latestWorkloadSnapshot?: () => WorkloadSnapshot | undefined;
} = {}
) {
let clock = 0;
Expand All @@ -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;
Expand All @@ -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');
Expand Down Expand Up @@ -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++;
Expand All @@ -156,6 +166,7 @@ function harness(
logs,
counts,
exits,
restartInfos,
unavailable: () => unavailable,
silence(ms = 30000) {
clock += ms;
Expand Down Expand Up @@ -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<string, number | boolean>;
};
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);
}
});
});
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -154,6 +159,7 @@ export type KiloRestartInfo = {
directory: string;
reason: KiloRestartReason;
interruptedExecutions?: ExecutionIdentity[];
outcomeReason?: string;
};

/**
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 =
Comment thread
eshurakov marked this conversation as resolved.
Resolved
restartWorkload !== undefined && isWorkloadAtCap(restartWorkload)
? ('sandbox_out_of_memory' as const)
: undefined;
let diagnostic = '';
if (reason !== 'credentials') {
const observation = trigger === 'health_probe_false' ? healthObservation : undefined;
Expand All @@ -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(
Expand Down Expand Up @@ -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=${
Expand Down
23 changes: 23 additions & 0 deletions services/cloud-agent-next/wrapper/src/control-plane/turn.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand Down
5 changes: 3 additions & 2 deletions services/cloud-agent-next/wrapper/src/control-plane/turn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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') {
Expand All @@ -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;
Expand Down Expand Up @@ -1036,7 +1037,7 @@ export function createTurnManager(deps: TurnManagerDeps) {
drainInbox(turn);
continue;
}
sendOutcome(turn, 'failed', 'agent_restarted');
sendOutcome(turn, 'failed', outcomeReason);
Comment thread
eshurakov marked this conversation as resolved.
Resolved
}
const reported = new Set<string>();
for (const execution of info.interruptedExecutions ?? []) {
Expand Down
Loading
Loading