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
21 changes: 21 additions & 0 deletions services/cloud-agent-next/src/sandbox-control/diagnostics.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,27 @@ describe('logControlDiagnostic', () => {
});
expect(withFields).toHaveBeenCalledTimes(1);
});

it('suppresses applied delta progress but logs a rejected delta session event', () => {
logControlDiagnostic(
'session_event_result',
{ eventType: 'message.part.delta', applied: true },
'info'
);
expect(withFields).not.toHaveBeenCalled();

logControlDiagnostic(
'session_event_result',
{ eventType: 'message.part.delta', applied: false },
'info'
);
expect(withFields).toHaveBeenCalledTimes(1);
expect(withFields.mock.calls[0]?.[0]).toMatchObject({
diagnosticEvent: 'session_event_result',
eventType: 'message.part.delta',
applied: false,
});
});
});

describe('diagnosticCause', () => {
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { afterEach, describe, expect, it, vi } from 'vitest';
import { SANDBOX_CONTROL_FORWARD_OPERATION_LIMIT } from '../shared/sandbox-control-protocol.js';
import { createSessionForwarding } from './session-forwarding.js';

describe('createSessionForwarding', () => {
Expand Down Expand Up @@ -118,6 +119,41 @@ describe('createSessionForwarding', () => {
expect(forwarding.stats()).toMatchObject({ waiting: 0, inFlight: 0, bufferedBytes: 0 });
});

it('admits up to the forward operation limit and rejects beyond it', async () => {
const forwarding = createSessionForwarding();
const release = Promise.withResolvers<void>();
const admitted: Array<Promise<unknown>> = [];
for (let index = 0; index < SANDBOX_CONTROL_FORWARD_OPERATION_LIMIT; index++) {
admitted.push(
forwarding.enqueueFenced({
sessionId: 'workspace_1',
bytes: 1,
deadlineAt: Date.now() + 60_000,
fence: async () => true,
forward: async () => release.promise,
})
);
}
expect(forwarding.stats()).toMatchObject({
waiting: SANDBOX_CONTROL_FORWARD_OPERATION_LIMIT,
inFlight: 0,
});

await expect(
forwarding.enqueueFenced({
sessionId: 'workspace_1',
bytes: 1,
deadlineAt: Date.now() + 60_000,
fence: async () => true,
forward: async () => 'unreachable',
})
).rejects.toMatchObject({ retryable: true });

release.resolve();
await Promise.allSettled(admitted);
expect(forwarding.stats()).toMatchObject({ waiting: 0, inFlight: 0, bufferedBytes: 0 });
});

it('keeps a detached lane until admitted work settles', async () => {
const forwarding = createSessionForwarding();
const release = Promise.withResolvers<void>();
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import {
MAX_SANDBOX_CONTROL_FRAME_BYTES,
SANDBOX_CONTROL_OPERATION_LIMIT,
SANDBOX_CONTROL_FORWARD_OPERATION_LIMIT,
} from '../shared/sandbox-control-protocol.js';

const MAX_SESSION_FORWARD_BYTES = 4 * MAX_SANDBOX_CONTROL_FRAME_BYTES;
Expand Down Expand Up @@ -88,7 +88,7 @@ export function createSessionForwarding(): SessionForwarding {
if (input.bytes > MAX_SANDBOX_CONTROL_FRAME_BYTES)
return Promise.reject(new SessionForwardingError('Forwarded frame is too large', false));
if (
stats.waiting + stats.inFlight >= SANDBOX_CONTROL_OPERATION_LIMIT ||
stats.waiting + stats.inFlight >= SANDBOX_CONTROL_FORWARD_OPERATION_LIMIT ||
stats.bufferedBytes + input.bytes > MAX_SESSION_FORWARD_BYTES
)
return Promise.reject(
Expand Down
35 changes: 26 additions & 9 deletions services/cloud-agent-next/src/sandbox-session/SandboxSession.ts
Original file line number Diff line number Diff line change
Expand Up @@ -644,7 +644,8 @@ export class SandboxSession extends DurableObject<Env> {
if (input.payload.type === 'session.created' || input.payload.type === 'session.updated') {
const info = input.payload.properties.info;
if (typeof info === 'object' && info !== null) {
if ('id' in info && info.id !== eventKiloSessionId) return { applied: false };
if ('id' in info && info.id !== eventKiloSessionId)
return result(false, 'session_id_mismatch');
if (eventKiloSessionId !== root && ('parentID' in info || 'directory' in info)) {
const directory = this.directory(metadata);
const child = childSessionLineage(info, directory);
Expand All @@ -653,7 +654,7 @@ export class SandboxSession extends DurableObject<Env> {
child.sessionId !== eventKiloSessionId ||
input.identity.directory !== directory
)
return { applied: false };
return result(false, 'child_lineage_mismatch');
}
}
}
Expand Down Expand Up @@ -949,12 +950,7 @@ export class SandboxSession extends DurableObject<Env> {
wrapperInstanceId?: string;
}): Promise<SandboxEventBatchResult> {
const outcomes: SandboxEventBatchItemOutcome[] = [];
let halted = false;
for (const item of input.items) {
if (halted) {
outcomes.push({ receiptId: item.receiptId, status: 'unattempted' });
continue;
}
for (const [index, item] of input.items.entries()) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bot: Continuing after an unknown item can apply same-session events out of order.

Suggested fix: Restore the halt guard after the catch: leave the throwing item as unknown, mark every remaining item unattempted, and emit one bounded session_event_batch_halted diagnostic for the failed item plus suffix shape. This keeps later deltas/snapshots/completions from being applied after a predecessor whose mutation outcome is indeterminate. The changed test at session-message-queue.test.ts:2226-2275 should again assert that the suffix is unattempted rather than applied.

try {
let applied = false;
let retryable: boolean | undefined;
Expand Down Expand Up @@ -990,7 +986,28 @@ export class SandboxSession extends DurableObject<Env> {
);
} catch {
outcomes.push({ receiptId: item.receiptId, status: 'unknown' });
halted = true;
try {
logControlDiagnostic(
'session_event_batch_item_failed',
{
sessionId: this.sessionId,
wrapperInstanceId: input.wrapperInstanceId,
receiptId: item.receiptId,
sequence: item.sequence,
eventFamily: item.event,
eventType:
item.event === 'session.event'
? diagnosticEventType(item.payload.type)
: 'session.preparing',
eventIndex: index,
batchSize: input.items.length,
disposition: 'application_exception',
},
'warn'
);
} catch {
// Containment only: a diagnostic must never replace or alter the batch result.
}
}
}
return { outcomes };
Expand Down
Loading