test/cascade-turn-deadline.test.js
Rama master · réplica de lectura del repositorio original.
// F3 (evidence/audit2/Cascade.md:280-296): the route budget did not cover the
// send phase. chat.js passes `maxWaitMs: budget.remainingMs()` — a DURATION —
// and the producer started its clock only after warmup + StartCascade + up to
// four Sends. Measured: a 100 ms budget took 310 ms behind a 250 ms Send, and
// the poll loop then received the full remainder again.
//
// The contract pinned here:
// * ONE absolute turn deadline is captured at producer entry and every phase
// (warmup, StartCascade, Send/retry, polling, detach admission) spends
// from it;
// * a slow Send cannot outlive the budget and the turn ends as a budget
// stop, not as an idle/transport verdict;
// * no new upstream attempt is started after the deadline (a timed-out
// recovery StartCascade does not lead to a second Send);
// * controls: a fast Send, a retried Send inside the budget, and a long
// valid multi-poll answer still complete normally.
//
// The seam is the real `cascadeChat` against a fake HTTP/2 language server,
// with arrival timestamps recorded server-side so "after the deadline" is a
// measured fact rather than an inference from elapsed time.
import './setup-env.mjs';
import { afterEach, describe, it } from 'node:test';
import assert from 'node:assert/strict';
import http2 from 'node:http2';
import { terminateRpcWithStatus } from './helpers/fake-grpc-trailers.mjs';
// Product imports are dynamic and come AFTER this block: src/client.js freezes
// CASCADE_TIMEOUTS at module load.
Object.assign(process.env, {
GRPC_PROTOCOL: 'connect',
CASCADE_POLL_INTERVAL_MS: '10',
CASCADE_IDLE_GRACE_MS: '1',
CASCADE_MAX_WAIT_MS: '1500',
CASCADE_COLD_STALL_BASE_MS: '1500',
CASCADE_WARM_STALL_MS: '1500',
WINDSURFAPI_USER_JWT: '0',
RESPONSE_CACHE_ENABLED: '0',
});
const { unwrapRequest, wrapEnvelope, endOfStreamEnvelope } = await import('../src/connect.js');
const { writeMessageField, writeStringField, writeVarintField } = await import('../src/proto.js');
const { isBudgetExceededError } = await import('../src/request-budget.js');
const { WindsurfClient } = await import('../src/client.js');
const MODEL_UID = 'claude-sonnet-4-6';
const MESSAGES = [{ role: 'user', content: 'hi' }];
afterEach(() => {
delete process.env.WINDSURFAPI_CASCADE_TURN;
});
function sleep(ms) {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function plannerStepBody({ responseText = '', modifiedText = '', status = 3 }) {
const fields = [];
if (responseText) fields.push(writeStringField(1, responseText));
if (modifiedText) fields.push(writeStringField(8, modifiedText));
return Buffer.concat([
writeVarintField(1, 15),
writeVarintField(4, status),
writeMessageField(20, Buffer.concat(fields)),
]);
}
function stepsBody(texts) {
return Buffer.concat(texts.map((text) => writeMessageField(1, plannerStepBody({ responseText: text }))));
}
function connectErrorBody(message) {
const bytes = Buffer.from(JSON.stringify({ error: { message } }));
const frame = Buffer.alloc(5 + bytes.length);
frame[0] = 2;
frame.writeUInt32BE(bytes.length, 1);
bytes.copy(frame, 5);
return frame;
}
/**
* Fake LS. `plan(method, state, payload)` returns
* `{ delayMs, body, error }`; arrival timestamps are recorded before the delay.
*/
async function withFakeLs(plan, fn) {
const requests = [];
const state = { starts: 0, sends: 0, stepReads: 0, statusReads: 0 };
const server = http2.createServer();
const sessions = new Set();
server.on('session', (session) => {
sessions.add(session);
session.on('close', () => sessions.delete(session));
});
server.on('stream', (stream, headers) => {
terminateRpcWithStatus(stream);
stream.on('error', () => {});
const chunks = [];
stream.on('data', (chunk) => chunks.push(chunk));
stream.on('end', () => {
let payload = Buffer.alloc(0);
try { payload = unwrapRequest(Buffer.concat(chunks), headers); } catch {}
const method = String(headers[':path'] || '').split('/').pop();
requests.push({ method, at: Date.now() });
if (method === 'StartCascade') state.starts++;
if (method === 'SendUserCascadeMessage') state.sends++;
if (method === 'GetCascadeTrajectorySteps') state.stepReads++;
if (method === 'GetCascadeTrajectory') state.statusReads++;
const step = plan(method, state, payload) || {};
const send = () => {
// The client may have cancelled at the deadline; a late response is
// exactly what this fixture must tolerate.
try {
stream.respond({
':status': 200,
'content-type': headers['content-type'] || 'application/connect+proto',
});
const body = step.error ? connectErrorBody(step.error) : (step.body || Buffer.alloc(0));
stream.end(step.error ? body : Buffer.concat([wrapEnvelope(body, { compress: false }), endOfStreamEnvelope()]));
} catch {}
};
if (step.delayMs > 0) setTimeout(send, step.delayMs);
else send();
});
});
await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
const port = server.address().port;
try {
return await fn({ port, requests, state });
} finally {
for (const session of sessions) {
try { session.close(); } catch {}
try { session.destroy(); } catch {}
}
await new Promise((resolve) => server.close(resolve));
}
}
// The deadline contract starts at producer entry, so the per-LS workspace
// handshake is not what these arms measure. Skipping it keeps the budget
// arithmetic deterministic on a slow checkout (the real warmup has its own
// never-finishes arm below).
class NoWarmClient extends WindsurfClient {
warmupCascade() { return Promise.resolve(); }
}
async function runTurn({ port, maxWaitMs, onChunk, signal, detach, Client = NoWarmClient, clockCapMs }) {
const client = new Client('test-api-key', port, 'csrf-token');
const opts = { onChunk, signal, detach };
if (maxWaitMs !== undefined) opts.maxWaitMs = maxWaitMs;
const realNow = Date.now;
let clockAnchor = null;
// Test seam for the wall-clock/timer rounding boundary: once the turn's
// first clock read happens, the wall clock is capped at `clockCapMs` from
// there while the real timers keep running. This reproduces a deadline
// timer that fires with a millisecond or two still reading as "left",
// deterministically instead of waiting for a loaded scheduler to do it.
if (clockCapMs !== undefined) {
Date.now = () => {
const now = realNow();
if (clockAnchor === null) clockAnchor = now;
return Math.min(now, clockAnchor + clockCapMs);
};
}
const started = realNow();
try {
const chunks = await client.cascadeChat(MESSAGES, 0, MODEL_UID, opts);
return {
started,
elapsed: realNow() - started,
chunks,
text: chunks.map((chunk) => chunk.text || '').join(''),
};
} catch (err) {
return { started, elapsed: realNow() - started, err };
} finally {
if (clockCapMs !== undefined) Date.now = realNow;
}
}
function idleStatus() {
return writeVarintField(2, 1);
}
describe('F3 the route budget covers the whole turn', () => {
it('a slow Send cannot outlive the budget, and nothing is polled after it', async () => {
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { body: writeStringField(1, `cascade-${state.starts}`) };
if (method === 'SendUserCascadeMessage') return { delayMs: 400 };
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const budgetMs = 120;
const result = await runTurn({ port: server.port, maxWaitMs: budgetMs });
assert.ok(result.elapsed < 400,
`a ${budgetMs}ms budget took ${result.elapsed}ms behind a 400ms Send`);
assert.ok(isBudgetExceededError(result.err),
`the turn must end as a budget stop, got: ${result.err?.message || JSON.stringify(result.text)}`);
assert.equal(result.err.type, 'upstream_deadline_exceeded');
const deadline = result.started + budgetMs;
const late = server.requests.filter((request) => request.at > deadline + 60);
assert.deepEqual(late.map((request) => request.method), [],
'no upstream attempt may be issued after the deadline');
assert.equal(server.state.sends, 1, 'the timed-out Send is not retried after the deadline');
assert.equal(server.state.stepReads, 0, 'the poll loop must not run on an exhausted budget');
assert.equal(server.state.statusReads, 0);
});
});
it('a StartCascade that outlives the budget stops the turn before Send', async () => {
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { delayMs: 400, body: writeStringField(1, `cascade-${state.starts}`) };
if (method === 'SendUserCascadeMessage') return {};
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const budgetMs = 120;
const result = await runTurn({ port: server.port, maxWaitMs: budgetMs });
assert.ok(result.elapsed < 400, `StartCascade outlived the budget: ${result.elapsed}ms`);
assert.ok(isBudgetExceededError(result.err), `expected a budget stop, got: ${result.err?.message}`);
assert.equal(server.state.starts, 1);
assert.equal(server.state.sends, 0, 'no Send may follow an expired StartCascade');
assert.equal(server.state.stepReads, 0);
});
});
it('a phase timer that fires on the last budget tick is still the budget stop', async () => {
// B06 CI flake (release attempt 2): the same StartCascade case failed with
// "expected a budget stop, got: gRPC unary timeout" after 120.24 ms. The
// StartCascade timer is armed from an integer-millisecond wall clock; when
// it fires, Date.now() can still read 1-2 ms of budget left because both
// endpoints truncate. `throwIfSpent` treated that as "not spent" and
// rethrew a transport error. Pin the boundary deterministically by capping
// the wall clock one millisecond short of the budget while the real timer
// runs its full remaining slice.
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { delayMs: 400, body: writeStringField(1, `cascade-${state.starts}`) };
if (method === 'SendUserCascadeMessage') return {};
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const budgetMs = 120;
const result = await runTurn({ port: server.port, maxWaitMs: budgetMs, clockCapMs: budgetMs - 1 });
assert.ok(result.elapsed < 400, `the capped clock stretched the turn: ${result.elapsed}ms`);
assert.ok(isBudgetExceededError(result.err),
`a timer firing on the last budget tick must be the budget stop, got: ${result.err?.message}`);
assert.equal(result.err.type, 'upstream_deadline_exceeded');
assert.equal(server.state.starts, 1);
assert.equal(server.state.sends, 0, 'no Send may follow the expired StartCascade');
assert.equal(server.state.stepReads, 0);
});
});
it('a warmup that never finishes is bounded by the deadline and starts no RPC', async () => {
class NeverWarmClient extends WindsurfClient {
warmupCascade() { return new Promise(() => {}); }
}
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { body: writeStringField(1, `cascade-${state.starts}`) };
if (method === 'SendUserCascadeMessage') return {};
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const budgetMs = 120;
const outcome = await Promise.race([
runTurn({ port: server.port, maxWaitMs: budgetMs, Client: NeverWarmClient }),
sleep(1500).then(() => ({ hung: true })),
]);
assert.equal(outcome.hung, undefined, 'the warmup await is outside the turn deadline');
assert.ok(isBudgetExceededError(outcome.err), `expected a budget stop, got: ${outcome.err?.message}`);
assert.ok(outcome.elapsed < 600, `warmup outlived the budget: ${outcome.elapsed}ms`);
assert.deepEqual(server.requests.map((request) => request.method), [],
'an exhausted warmup must not be followed by any upstream RPC');
});
});
it('a recovery StartCascade cannot outlive the deadline and is not followed by a second Send', async () => {
await withFakeLs((method, state) => {
if (method === 'StartCascade') {
return state.starts === 1
? { body: writeStringField(1, 'cascade-1') }
: { delayMs: 400, body: writeStringField(1, `cascade-${state.starts}`) };
}
if (method === 'SendUserCascadeMessage') {
return state.sends === 1 ? { error: 'panel state not found for sessionId fixture' } : {};
}
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const budgetMs = 150;
const result = await runTurn({ port: server.port, maxWaitMs: budgetMs });
assert.ok(result.elapsed < 400,
`the recovery StartCascade outlived the budget: ${result.elapsed}ms`);
assert.ok(isBudgetExceededError(result.err), `expected a budget stop, got: ${result.err?.message}`);
assert.equal(server.state.sends, 1, 'no second Send after the deadline');
assert.equal(server.state.starts, 2, 'the recovery StartCascade was attempted inside the budget');
assert.equal(server.state.stepReads, 0);
});
});
it('the poll window is what the send phase left, not a fresh copy of the budget', async () => {
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { body: writeStringField(1, `cascade-${state.starts}`) };
// 120 ms of a 200 ms budget is spent before polling begins.
if (method === 'SendUserCascadeMessage') return { delayMs: 120 };
// The trajectory stays ACTIVE with no text: only the window's own length
// decides when this turn ends.
if (method === 'GetCascadeTrajectorySteps') return {};
if (method === 'GetCascadeTrajectory') return { body: writeVarintField(2, 2) };
return {};
}, async (server) => {
const result = await runTurn({ port: server.port, maxWaitMs: 200 });
assert.ok(result.elapsed < 340,
`the poll loop re-ran the full budget after the send phase: ${result.elapsed}ms for a 200ms turn`);
assert.ok(server.state.statusReads < 15,
`the poll loop ran ${server.state.statusReads} status reads — a fresh 200ms window instead of the send-phase remainder`);
assert.ok(result.elapsed >= 180,
`the turn ended before its deadline: ${result.elapsed}ms`);
assert.equal(result.err, undefined, `a window expiry is a typed terminal, not a throw: ${result.err?.message}`);
assert.equal(result.chunks.endReason, 'max_wait');
assert.equal(result.chunks.terminal.kind, 'empty');
assert.equal(result.chunks.terminal.status, 504);
assert.equal(result.chunks.terminal.code, 'windsurf_provider_deadline');
});
});
it('control: a fast Send inside the budget completes normally', async () => {
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { body: writeStringField(1, `cascade-${state.starts}`) };
if (method === 'SendUserCascadeMessage') return {};
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const result = await runTurn({ port: server.port, maxWaitMs: 2000 });
assert.equal(result.err, undefined, `turn failed: ${result.err?.message}`);
assert.equal(result.text, 'answer');
assert.equal(result.chunks.endReason, 'idle_done');
assert.equal(result.chunks.terminal.ok, true);
assert.ok(server.state.stepReads >= 1, 'the turn actually polled');
assert.ok(result.elapsed < 2000);
});
});
it('control: a retried Send inside the budget still recovers', async () => {
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { body: writeStringField(1, `cascade-${state.starts}`) };
if (method === 'SendUserCascadeMessage') {
return state.sends === 1 ? { error: 'panel state not found for sessionId fixture' } : {};
}
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['recovered-answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const result = await runTurn({ port: server.port, maxWaitMs: 3000 });
assert.equal(result.err, undefined, `turn failed: ${result.err?.message}`);
assert.equal(server.state.sends, 2, 'the panel-missing Send is retried once');
assert.equal(server.state.starts, 2, 'the recovery opened a fresh cascade');
assert.equal(result.text, 'recovered-answer');
assert.equal(result.chunks.endReason, 'idle_done');
});
});
it('control: a long valid multi-poll answer is not cut short by the deadline', async () => {
const block = 'L'.repeat(8000);
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { body: writeStringField(1, `cascade-${state.starts}`) };
if (method === 'SendUserCascadeMessage') return {};
if (method === 'GetCascadeTrajectorySteps') {
const read = state.stepReads;
if (read === 1) return { body: stepsBody([block]) };
if (read === 2) return { body: stepsBody([block, block]) };
return { body: stepsBody([block, block, block]) };
}
if (method === 'GetCascadeTrajectory') {
// ACTIVE while the answer is still growing, IDLE once it has settled.
return { body: writeVarintField(2, state.stepReads <= 3 ? 2 : 1) };
}
return {};
}, async (server) => {
const result = await runTurn({ port: server.port });
assert.equal(result.err, undefined, `turn failed: ${result.err?.message}`);
assert.equal(result.text, block.repeat(3), 'every byte of a long answer is delivered once, in order');
assert.equal(result.chunks.totalYielded, block.length * 3);
assert.equal(result.chunks.endReason, 'idle_done');
assert.equal(result.chunks.terminal.ok, true);
assert.ok(result.elapsed > 30, 'the control really spanned several polls');
});
});
it('detach admission is pinned to the absolute turn deadline, not the poll-window start', async () => {
process.env.WINDSURFAPI_CASCADE_TURN = '1';
const pins = [];
await withFakeLs((method, state) => {
if (method === 'StartCascade') return { body: writeStringField(1, `cascade-${state.starts}`) };
// The Send spends part of the budget before the poll phase begins.
if (method === 'SendUserCascadeMessage') return { delayMs: 150 };
if (method === 'GetCascadeTrajectorySteps') return { body: stepsBody(['answer']) };
if (method === 'GetCascadeTrajectory') return { body: idleStatus() };
return {};
}, async (server) => {
const budgetMs = 600;
const controller = new AbortController();
const result = await runTurn({
port: server.port,
maxWaitMs: budgetMs,
signal: controller.signal,
detach: {
forcedCause: null,
hasToolCalls: () => false,
admit: (snapshot, pin) => { pins.push(pin); },
},
onChunk: (chunk) => { if (chunk.text) controller.abort(); },
});
assert.equal(result.err?.name, 'AbortError', `expected the abort contract, got: ${result.err?.message || 'a result'}`);
assert.equal(pins.length, 1, 'the turn was admitted for detach exactly once');
const pin = pins[0];
assert.equal(pin.deadline - pin.startedAt, budgetMs,
'the admitted claim owns the turn for exactly the route budget');
assert.ok(pin.deadline >= result.started + budgetMs - 20,
`the claim deadline is earlier than the turn deadline: ${pin.deadline - result.started}ms`);
assert.ok(pin.deadline <= result.started + budgetMs + 100,
`the claim deadline includes the send phase: ${pin.deadline - result.started}ms for a ${budgetMs}ms budget`);
});
});
});