test/detach-soak.test.js
Rama master · réplica de lectura del repositorio original.
// Detach P1 — end-to-end lazy resume (T1–T19).
//
// Harness: a scripted loopback LS (the shape used by
// test/cascade-abort-side-effects.test.js) plus a response double that closes
// the socket at a chosen write. The producer is the real cascadeChat; the
// handler is the real streamResponse.
//
// What P1 promises and these tests pin:
// - an aborted mid-stream turn freezes a v2 snapshot (T1);
// - a byte-identical retry CLAIMS it and continues the original poll loop:
// no StartCascade, no SendUserCascadeMessage, prefix replayed exactly once
// (T5, T7, T11);
// - billing starts the metadata RPC at the snapshot's generatorOffset (T3)
// and lands on the entry's owner account even though the resume has no
// checked-out account (T5);
// - every negative (pre-content, tool call, tenant, shutdown, drain, TTL,
// generation, cascade gone, account removed) admits nothing and falls
// back to upstream, each with a positive control in the same file.
import { after, before, beforeEach, describe, it } from 'node:test';
import assert from 'node:assert/strict';
import http2 from 'node:http2';
import { terminateRpcWithStatus, isConnectContentType } from './helpers/fake-grpc-trailers.mjs';
import { endOfStreamEnvelope, unwrapRequest, wrapEnvelope } from '../src/connect.js';
import { getField, parseFields, writeMessageField, writeStringField, writeVarintField } from '../src/proto.js';
// Module-level env (client.js snapshots CASCADE_TIMEOUTS, grpc.js snapshots
// GRPC_PROTOCOL). Both switches are on for this file; the flag-off parity test
// deletes the detach switch around its request.
process.env.GRPC_PROTOCOL = 'connect';
process.env.CASCADE_POLL_INTERVAL_MS = '5';
process.env.CASCADE_IDLE_GRACE_MS = '1';
process.env.CASCADE_MAX_WAIT_MS = '200000';
process.env.CASCADE_COLD_STALL_BASE_MS = '200000';
process.env.CASCADE_WARM_STALL_MS = '200000';
process.env.RESPONSE_CACHE_ENABLED = '0';
process.env.WINDSURFAPI_CASCADE_TURN = '1';
process.env.WINDSURFAPI_CASCADE_DETACH = '1';
const PREFIX = 'PREFIX-ANSWER-';
const GROWTH = 'GROWTH-AFTER-DETACH-';
const FULL_TEXT = `${PREFIX}${GROWTH}`;
let WindsurfClient;
let handleChatCompletions;
let resolveModel;
let auth;
let cache;
let pool;
let registry;
let sse;
// ── proto helpers ────────────────────────────────────────────
function grpcFrame(payload) {
const buf = Buffer.isBuffer(payload) ? payload : Buffer.from(payload);
const frame = Buffer.alloc(5 + buf.length);
frame[0] = 0;
frame.writeUInt32BE(buf.length, 1);
buf.copy(frame, 5);
return frame;
}
function responseBody(payload, headers) {
const contentType = String(headers['content-type'] || '');
if (isConnectContentType(contentType)) {
return Buffer.concat([wrapEnvelope(payload, { compress: false }), endOfStreamEnvelope()]);
}
return grpcFrame(payload);
}
function endOfStreamError(message, code = 'not_found') {
const trailer = Buffer.from(JSON.stringify({ error: { code, message } }));
const frame = Buffer.alloc(5 + trailer.length);
frame[0] = 0x02;
frame.writeUInt32BE(trailer.length, 1);
trailer.copy(frame, 5);
return frame;
}
function plannerStep(text, { status = 3, thinking = '', toolCall = null } = {}) {
const plannerParts = [writeStringField(1, text)];
if (thinking) plannerParts.push(writeStringField(3, thinking));
const parts = [
writeVarintField(1, 15), // CORTEX_STEP_TYPE_PLANNER_RESPONSE
writeVarintField(4, status),
writeMessageField(20, Buffer.concat(plannerParts)),
];
if (toolCall) {
const call = Buffer.concat([
writeStringField(1, toolCall.id),
writeStringField(2, toolCall.name),
writeStringField(3, toolCall.argumentsJson),
]);
// 49 = tool_call_proposal { 1 = ChatToolCall }
parts.push(writeMessageField(49, writeMessageField(1, call)));
}
return Buffer.concat(parts);
}
function rawStepsResponse(steps) {
return Buffer.concat(steps.map(step => writeMessageField(1, step)));
}
function readStepOffset(payload) {
const field = getField(parseFields(payload), 2, 0);
return field ? Number(field.value) : 0;
}
function metadataEntry(inputTokens, outputTokens) {
const usage = Buffer.concat([
writeVarintField(2, inputTokens),
writeVarintField(3, outputTokens),
]);
const chatModel = writeMessageField(4, usage);
return writeMessageField(1, chatModel); // CortexStepGeneratorMetadata.chat_model
}
function metadataResponse(records) {
return Buffer.concat(records.map(record => writeMessageField(1, record)));
}
// ── scripted upstream ────────────────────────────────────────
/**
* One fake LS per test. `state.grown` flips the same step from PREFIX to
* FULL_TEXT (in-place growth, the E2b shape). `state.status` is ACTIVE until
* the test flips it to IDLE. `state.failNextSteps` makes the next steps read
* a Connect end-stream error (cascade gone).
*/
function cascadeUpstream({ initialText = PREFIX, toolCall = null, thinkingOnly = false } = {}) {
const state = {
startCalls: 0,
sendCalls: 0,
stepsCalls: 0,
statusCalls: 0,
metadataCalls: 0,
metadataOffsets: [],
stepsOffsets: [],
grown: false,
status: 'active',
failNextSteps: 0,
resumeDelayMs: 0,
firstResumeRead: true,
stepsFor: null, // optional (offset) => Buffer
};
const handler = (stream, headers) => {
const chunks = [];
stream.on('data', chunk => chunks.push(chunk));
stream.on('end', () => {
const method = String(headers[':path'] || '').split('/').pop();
const contentType = headers['content-type'] || 'application/grpc';
const payload = unwrapRequest(Buffer.concat(chunks), headers);
const respond = (body) => {
stream.respond({ ':status': 200, 'content-type': contentType });
stream.end(responseBody(body, headers));
};
if (method === 'StartCascade') {
state.startCalls += 1;
respond(writeStringField(1, 'detach-cascade'));
return;
}
if (method === 'SendUserCascadeMessage') {
state.sendCalls += 1;
respond(Buffer.alloc(0));
return;
}
if (method === 'GetCascadeTrajectorySteps') {
state.stepsCalls += 1;
const offset = readStepOffset(payload);
state.stepsOffsets.push(offset);
if (state.failNextSteps > 0) {
state.failNextSteps -= 1;
stream.respond({ ':status': 200, 'content-type': 'application/connect+proto' });
stream.end(endOfStreamError('cascade not found'));
return;
}
const delay = state.grown && state.firstResumeRead ? state.resumeDelayMs : 0;
state.firstResumeRead = false;
const bodyFor = () => {
if (state.stepsFor) return state.stepsFor(offset);
const text = state.grown ? (thinkingOnly ? '' : FULL_TEXT) : initialText;
const step = plannerStep(text, {
status: state.grown ? 3 : 1,
thinking: thinkingOnly ? (state.grown ? `${PREFIX}${GROWTH}` : PREFIX) : '',
toolCall: state.grown ? null : toolCall,
});
return rawStepsResponse(offset === 0 ? [step] : []);
};
if (delay > 0) {
setTimeout(() => respond(bodyFor()), delay);
} else {
respond(bodyFor());
}
return;
}
if (method === 'GetCascadeTrajectory') {
state.statusCalls += 1;
respond(writeVarintField(2, state.status === 'idle' ? 1 : 2));
return;
}
if (method === 'GetCascadeTrajectoryGeneratorMetadata') {
state.metadataCalls += 1;
const offset = readStepOffset(payload);
state.metadataOffsets.push(offset);
// Offset 0 reports THREE records (the whole cascade history); offset 7
// reports ONE (the current turn only). A resume that forgets the
// snapshot offset over-bills by the difference.
respond(metadataResponse(offset === 0
? [metadataEntry(100, 10), metadataEntry(100, 10), metadataEntry(100, 10)]
: [metadataEntry(100, 10)]));
return;
}
respond(Buffer.alloc(0));
});
};
return { state, handler };
}
async function withFakeLanguageServer(handler, fn) {
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); handler(stream, headers); });
await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
const port = server.address().port;
try {
return await fn(port);
} finally {
for (const session of sessions) {
try { session.close(); } catch {}
try { session.destroy(); } catch {}
}
await new Promise(resolve => server.close(resolve));
}
}
// ── response doubles ────────────────────────────────────────
/**
* A response with Node's destroyed-peer facts: `close` fires, writableEnded
* stays false, writes after close are accepted silently. `closeAfterWrites`
* fires the close AFTER the Nth write is recorded.
*/
function makeResponse({ closeAfterWrites = 0, backpressure = false } = {}) {
const listeners = new Map();
const writes = [];
const state = { closed: false, writableEnded: false, statusCode: 0, body: '' };
const res = {
statusCode: 0,
get writableEnded() { return state.writableEnded; },
writeHead(status) { state.statusCode = status; res.statusCode = status; },
write(chunk) {
const text = String(chunk);
state.body += text;
writes.push({ text, afterClose: state.closed });
if (backpressure && text.includes('"content":"')) return false;
if (closeAfterWrites && writes.length >= closeAfterWrites && !state.closed) {
res.emit('close');
}
return true;
},
end(chunk) {
if (chunk) res.write(chunk);
state.writableEnded = true;
},
on(event, cb) {
if (!listeners.has(event)) listeners.set(event, []);
listeners.get(event).push(cb);
},
once(event, cb) {
const wrap = (...args) => { res.removeListener(event, wrap); cb(...args); };
res.on(event, wrap);
},
removeListener(event, cb) {
const list = listeners.get(event) || [];
const i = list.indexOf(cb);
if (i >= 0) list.splice(i, 1);
},
emit(event, ...args) {
if (event === 'close') state.closed = true;
for (const cb of [...(listeners.get(event) || [])]) cb(...args);
},
};
return { res, writes, state };
}
// ── handler context + accounts ──────────────────────────────
function seedAccount(label) {
const a = auth.addAccountByKey(`devin-session-token$xk-detach-${label}-${Math.random().toString(36).slice(2)}`, label);
const acct = auth.getAccountInternal(a.id);
acct.tier = 'pro';
acct.status = 'active';
return acct;
}
function makeLsFake(port) {
return {
port,
csrfToken: 'csrf',
generation: 1,
activeRequests: 0,
maxActive: 0,
beginCalls: 0,
endCalls: 0,
acquireCalls: 0,
};
}
function handlerContext(port, acct, lsFake, overrides = {}) {
return {
callerKey: 'session:detach-lazy-resume',
waitForAccount: (tried, _signal, _maxWait, modelKey) => (acct
? { id: acct.id, email: acct.email, apiKey: acct.apiKey, proxy: null }
: auth.getApiKey(tried, modelKey)),
ensureLs: async () => {},
getLsFor: () => ({ port, csrfToken: 'csrf', generation: lsFake ? lsFake.generation : 1 }),
getLsEntryByPort: (p) => (lsFake && p === port ? lsFake : null),
beginLsUse: (p) => {
if (!lsFake || p !== port) return null;
lsFake.beginCalls += 1;
lsFake.activeRequests += 1;
lsFake.maxActive = Math.max(lsFake.maxActive, lsFake.activeRequests);
return lsFake;
},
endLsUse: (p) => {
if (!lsFake || p !== port) return;
lsFake.endCalls += 1;
lsFake.activeRequests = Math.max(0, lsFake.activeRequests - 1);
},
acquireLsUse: (p) => {
if (!lsFake || p !== port) return null;
lsFake.acquireCalls += 1;
if (lsFake.activeRequests > 0) throw new Error('resume acquired a second LS lease');
lsFake.activeRequests += 1;
lsFake.maxActive = Math.max(lsFake.maxActive, lsFake.activeRequests);
return lsFake;
},
WindsurfClient,
...overrides,
};
}
function bodyFor(marker) {
return {
model: 'gemini-2.5-flash',
messages: [{ role: 'user', content: `detach test ${marker}` }],
stream: true,
};
}
const createdIds = [];
before(async () => {
({ WindsurfClient } = await import('../src/client.js'));
({ handleChatCompletions } = await import('../src/handlers/chat.js'));
({ resolveModel } = await import('../src/models.js'));
auth = await import('../src/auth.js');
cache = await import('../src/cache.js');
pool = await import('../src/conversation-pool.js');
registry = await import('../src/detached-registry.js');
sse = await import('../src/sse-registry.js');
});
beforeEach(() => {
cache.cacheClear();
pool.poolClear();
registry.detachedRegistry.reset();
});
after(() => {
registry.detachedRegistry.reset();
while (createdIds.length) {
try { auth.removeAccount(createdIds.pop()); } catch {}
}
cache.cacheClear();
pool.poolClear();
});
// ── drivers ─────────────────────────────────────────────────
async function startStream(body, ctx) {
const result = await handleChatCompletions(body, ctx);
assert.equal(result.status, 200, `expected a stream, got ${JSON.stringify(result.body || result)}`);
assert.equal(result.stream, true);
return result;
}
async function runToCompletion(body, ctx, responseOpts) {
const result = await startStream(body, ctx);
const { res, writes, state } = makeResponse(responseOpts);
await result.handler(res);
return { res, writes, state };
}
function detachKeyOf(body, callerKey = 'session:detach-lazy-resume') {
return registry.detachedKey(body, callerKey, {
scopeTrusted: true,
route: 'chat',
modelKey: resolveModel(body.model),
toolPreambleTier: null,
emulateTools: false,
native: null,
stop: [],
displayModel: body.model,
});
}
function contentOf(state) {
return state.body
.split('\n\n')
.filter(frame => frame.startsWith('data: '))
.map(frame => {
try { return JSON.parse(frame.slice(6)); } catch { return null; }
})
.filter(Boolean)
.map(data => data.choices?.[0]?.delta?.content || '')
.join('');
}
function spendOf(acct) {
return auth.getAccountPublic(acct.id)?.totalSpend?.totalTokens ?? 0;
}
async function waitFor(predicate, timeoutMs = 1000) {
const start = Date.now();
while (Date.now() - start < timeoutMs) {
if (predicate()) return true;
await new Promise(resolve => setTimeout(resolve, 2));
}
return false;
}
// ── soak scenarios ──────────────────────────────────────────
function snap() { return registry.detachedRegistry.stats(); }
// Opt-in: SOAK-2 waits the real 120 s TTL, so the whole file is ~2 min and the
// shard runner cancels it at 90 s. The unit suites pin the same behaviour with
// fake timers; run this one deliberately:
// DETACH_SOAK=1 node --import ./test/setup-env.mjs --test test/detach-soak.test.js
const SOAK_ENABLED = process.env.DETACH_SOAK === '1';
describe('Detach P1 soak', { skip: SOAK_ENABLED ? false : 'set DETACH_SOAK=1 (real 120 s TTL)' }, () => {
it('SOAK-1 ten detach/resume cycles: claim, replay, lease and bill stay exact', async () => {
const acct = seedAccount('soak1');
createdIds.push(acct.id);
const { state, handler } = cascadeUpstream();
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
for (let i = 0; i < 10; i++) {
state.grown = false;
state.status = 'active';
state.firstResumeRead = true;
const body = bodyFor(`soak1-${i}`);
const ctx = handlerContext(port, acct, lsFake);
const s0 = snap();
const starts0 = state.startCalls;
const sends0 = state.sendCalls;
const spend0 = spendOf(acct);
await runToCompletion(body, ctx, { closeAfterWrites: 2 });
assert.equal(snap().admit - s0.admit, 1, `cycle ${i}: admit`);
assert.equal(lsFake.activeRequests, 0, `cycle ${i}: detach pins no lease`);
state.status = 'idle';
state.grown = true;
const { state: resState } = await runToCompletion(body, ctx, {});
assert.equal(snap().serve - s0.serve, 1, `cycle ${i}: served`);
assert.equal(snap().entries, 0, `cycle ${i}: entry deleted`);
assert.equal(snap().bytes, 0, `cycle ${i}: bytes released`);
assert.equal(state.startCalls - starts0, 1, `cycle ${i}: resume did not StartCascade`);
assert.equal(state.sendCalls - sends0, 1, `cycle ${i}: resume did not SendUserCascadeMessage`);
assert.equal(spendOf(acct) - spend0, 330, `cycle ${i}: owner billed once (3-record metadata at offset 0)`);
assert.equal(resState.body.split(PREFIX).length - 1, 1, `cycle ${i}: prefix once`);
assert.equal(resState.body.split(GROWTH).length - 1, 1, `cycle ${i}: growth once`);
assert.equal(lsFake.activeRequests, 0, `cycle ${i}: lease released`);
assert.equal(lsFake.maxActive, 1, `cycle ${i}: exactly one lease`);
}
});
});
it('SOAK-2 the real 120 s timer evicts and the retry goes upstream', async () => {
const acct = seedAccount('soak2');
createdIds.push(acct.id);
const { state, handler } = cascadeUpstream();
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
const body = bodyFor('soak2');
const ctx = handlerContext(port, acct, lsFake);
await runToCompletion(body, ctx, { closeAfterWrites: 2 });
const key = detachKeyOf(body);
const entry = registry.detachedRegistry.lookup(key, ctx.callerKey);
assert.ok(entry, 'precondition: admitted');
const ttl0 = snap().evict.ttl;
const t0 = Date.now();
await new Promise(resolve => setTimeout(resolve, 121_000));
const waited = Date.now() - t0;
assert.ok(waited >= 120_000, `waited ${waited} ms`);
assert.equal(snap().entries, 0, 'the unref timer evicted the entry');
assert.equal(snap().evict.ttl - ttl0, 1, 'evict{ttl} counted once');
assert.equal(lsFake.activeRequests, 0, 'no lease pinned by the entry');
const starts0 = state.startCalls;
state.status = 'idle';
state.grown = true;
await runToCompletion(body, ctx, {});
assert.equal(state.startCalls - starts0, 1, 'the expired retry reaches upstream');
assert.equal(snap().serve, 0, 'nothing was resumed');
});
});
it('SOAK-3a per-caller cap evicts the oldest running entry; the evicted retry goes upstream', async () => {
const acct = seedAccount('soak3a');
createdIds.push(acct.id);
const { state, handler } = cascadeUpstream();
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
const ctx = handlerContext(port, acct, lsFake);
registry.detachedRegistry.maxPerCaller = 1;
try {
state.grown = false;
state.status = 'active';
state.firstResumeRead = true;
const bodyA = bodyFor('soak3a-a');
const bodyB = bodyFor('soak3a-b');
await runToCompletion(bodyA, ctx, { closeAfterWrites: 2 });
await runToCompletion(bodyB, ctx, { closeAfterWrites: 2 });
assert.equal(snap().admit, 2);
assert.equal(snap().evict.capacity, 1);
assert.equal(snap().entries, 1);
assert.equal(registry.detachedRegistry.lookup(detachKeyOf(bodyA), ctx.callerKey), null);
state.status = 'idle';
state.grown = true;
const starts0 = state.startCalls;
await runToCompletion(bodyA, ctx, {});
assert.equal(state.startCalls - starts0, 1, 'the evicted retry goes upstream');
state.firstResumeRead = true;
await runToCompletion(bodyB, ctx, {});
assert.equal(snap().serve, 1, 'the surviving entry still resumes');
assert.equal(snap().entries, 0);
assert.equal(lsFake.activeRequests, 0);
} finally {
registry.detachedRegistry.maxPerCaller = 2;
}
});
});
it('SOAK-3b entry cap evicts only running entries; a claimed entry is never dropped', async () => {
const acct = seedAccount('soak3b');
createdIds.push(acct.id);
const { state, handler } = cascadeUpstream();
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
const ctx = handlerContext(port, acct, lsFake);
registry.detachedRegistry.maxEntries = 1;
try {
state.grown = false;
state.status = 'active';
state.firstResumeRead = true;
await runToCompletion(bodyFor('soak3b-a'), ctx, { closeAfterWrites: 2 });
await runToCompletion(bodyFor('soak3b-b'), ctx, { closeAfterWrites: 2 });
assert.equal(snap().entries, 1);
assert.equal(snap().evict.capacity, 1);
} finally {
registry.detachedRegistry.maxEntries = 8;
}
const reg = new registry.DetachedRegistry({ maxEntries: 1, timers: false });
const snapOk = {
version: 2, stepOffset: 1, pollFloor: 0, yieldedByStep: [], thinkingByStep: [],
consumedThrough: 1, generatorOffset: 0, sawActive: false, usageByStep: [], chunks: [],
};
const pin = { deadline: Date.now() + 60_000 };
assert.ok(reg.admit({ key: 'claimed', callerKey: 'session:a', snapshot: snapOk, pin }).token);
assert.ok(reg.claim('claimed', 'session:a').entry);
assert.equal(
reg.admit({ key: 'next', callerKey: 'session:b', snapshot: snapOk, pin }).refused,
'over_entries',
'a claimed entry cannot be evicted to make room',
);
assert.ok(reg.lookup('claimed', 'session:a'), 'the claimed entry survives');
});
});
it('SOAK-3c byte cap refuses without truncation; the turn still falls back upstream', async () => {
const acct = seedAccount('soak3c');
createdIds.push(acct.id);
const big = cascadeUpstream({ initialText: 'X'.repeat(2000) });
await withFakeLanguageServer(big.handler, async (port) => {
const lsFake = makeLsFake(port);
const ctx = handlerContext(port, acct, lsFake);
registry.detachedRegistry.maxEntryBytes = 2048;
try {
const bytes0 = snap().refuse.over_bytes;
await runToCompletion(bodyFor('soak3c'), ctx, { closeAfterWrites: 2 });
assert.equal(snap().refuse.over_bytes - bytes0, 1, 'over_bytes refusal');
assert.equal(snap().entries, 0, 'no truncated entry');
assert.equal(lsFake.activeRequests, 0);
big.state.status = 'idle';
big.state.grown = true;
const starts0 = big.state.startCalls;
await runToCompletion(bodyFor('soak3c'), ctx, {});
assert.equal(big.state.startCalls - starts0, 1, 'the refused retry reaches upstream');
} finally {
registry.detachedRegistry.maxEntryBytes = 4 * 1024 * 1024;
}
});
});
it('SOAK-4 abort during prefix replay re-detaches three times without extending TTL', async () => {
const acct = seedAccount('soak4');
createdIds.push(acct.id);
const { state, handler } = cascadeUpstream();
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
for (let i = 0; i < 3; i++) {
state.grown = false;
state.status = 'active';
state.firstResumeRead = true;
const body = bodyFor(`soak4-${i}`);
const ctx = handlerContext(port, acct, lsFake);
await runToCompletion(body, ctx, { closeAfterWrites: 2 });
const key = detachKeyOf(body);
const originalExpiry = registry.detachedRegistry.lookup(key, ctx.callerKey).expiresAt;
state.status = 'idle';
state.grown = true;
const s0 = snap();
const spend0 = spendOf(acct);
await runToCompletion(body, ctx, { closeAfterWrites: 3 });
assert.equal(snap().re_detach - s0.re_detach, 1, `cycle ${i}: re-detach`);
assert.equal(snap().serve - s0.serve, 0, `cycle ${i}: not served`);
const re = registry.detachedRegistry.lookup(key, ctx.callerKey);
assert.ok(re, `cycle ${i}: entry is running again`);
assert.equal(re.expiresAt, originalExpiry, `cycle ${i}: TTL unchanged`);
assert.equal(
re.snapshot.chunks.map(chunk => chunk.text || '').join(''),
FULL_TEXT,
`cycle ${i}: snapshot has the full answer`,
);
assert.equal(spendOf(acct) - spend0, 0, `cycle ${i}: the aborted replay billed nothing`);
state.firstResumeRead = true;
const { state: finalState } = await runToCompletion(body, ctx, {});
assert.equal(snap().serve - s0.serve, 1, `cycle ${i}: the final retry is served`);
assert.equal(snap().entries, 0, `cycle ${i}: entry deleted`);
assert.equal(finalState.body.split(PREFIX).length - 1, 1, `cycle ${i}: prefix once`);
assert.equal(finalState.body.split(GROWTH).length - 1, 1, `cycle ${i}: growth once`);
assert.equal(spendOf(acct) - spend0, 330, `cycle ${i}: billed once`);
assert.equal(lsFake.activeRequests, 0, `cycle ${i}: lease released`);
}
});
});
it('SOAK-5 three claim races: one serve, one busy upstream, one detached bill', async () => {
const acct = seedAccount('soak5');
createdIds.push(acct.id);
const { state, handler } = cascadeUpstream();
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
for (let i = 0; i < 3; i++) {
state.grown = false;
state.status = 'active';
state.firstResumeRead = true;
const body = bodyFor(`soak5-${i}`);
const ctx = handlerContext(port, acct, lsFake);
await runToCompletion(body, ctx, { closeAfterWrites: 2 });
state.status = 'idle';
state.grown = true;
state.resumeDelayMs = 60;
const s0 = snap();
const starts0 = state.startCalls;
const first = await startStream(body, ctx);
const firstRes = makeResponse({});
const firstRun = first.handler(firstRes.res);
await new Promise(resolve => setTimeout(resolve, 15));
const second = await startStream(body, ctx);
const secondRes = makeResponse({});
await second.handler(secondRes.res);
await firstRun;
assert.equal(snap().claim.ok - s0.claim.ok, 1, `cycle ${i}: one claim`);
assert.equal(snap().claim.busy - s0.claim.busy, 1, `cycle ${i}: one busy`);
assert.equal(snap().serve - s0.serve, 1, `cycle ${i}: one serve`);
assert.equal(snap().billed_detached - s0.billed_detached, 1, `cycle ${i}: one detached bill`);
assert.equal(state.startCalls - starts0, 1, `cycle ${i}: the busy retry starts upstream`);
assert.ok(contentOf(firstRes.state).includes(GROWTH), `cycle ${i}: the resumed answer grew`);
assert.equal(contentOf(secondRes.state), FULL_TEXT, `cycle ${i}: the busy retry answers from upstream`);
assert.equal(lsFake.activeRequests, 0, `cycle ${i}: leases released`);
}
});
});
it('SOAK-6 a seeded multi-turn resume bills one record on the owner, not the retry account', async () => {
const owner = seedAccount('soak6-owner');
const other = seedAccount('soak6-other');
createdIds.push(owner.id, other.id);
const { state, handler } = cascadeUpstream();
state.stepsFor = (offset) => {
if (offset !== 7) return rawStepsResponse([]);
return rawStepsResponse([
plannerStep(state.grown ? FULL_TEXT : PREFIX, { status: state.grown ? 3 : 1 }),
]);
};
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
const body = {
model: 'gemini-2.5-flash',
messages: [
{ role: 'user', content: 'first turn' },
{ role: 'assistant', content: 'first answer' },
{ role: 'user', content: 'detach soak t6' },
],
stream: true,
};
const ownerCtx = handlerContext(port, owner, lsFake);
const modelKey = resolveModel(body.model);
const fpBefore = pool.fingerprintBefore(body.messages, modelKey, ownerCtx.callerKey, {
tools: [], toolChoice: null, toolPreamble: '', preambleTier: null,
emulateTools: false, route: 'chat',
});
pool.checkin(fpBefore, {
cascadeId: 'seeded-detach-soak6',
sessionId: 'seeded-session',
lsPort: port,
lsGeneration: 1,
apiKey: owner.apiKey,
modelKey,
stepOffset: 7,
generatorOffset: 7,
}, ownerCtx.callerKey, 0);
await runToCompletion(body, ownerCtx, { closeAfterWrites: 2 });
assert.equal(state.startCalls, 0, 'precondition: the original resumed the seeded cascade');
assert.equal(snap().admit, 1);
const entry = registry.detachedRegistry.lookup(detachKeyOf(body), ownerCtx.callerKey);
assert.equal(entry.snapshot.generatorOffset, 7);
assert.equal(entry.snapshot.stepOffset, 8);
state.status = 'idle';
state.grown = true;
state.metadataOffsets.length = 0;
const ownerBefore = spendOf(owner);
const otherBefore = spendOf(other);
const retryCtx = handlerContext(port, other, lsFake);
await runToCompletion(body, retryCtx, {});
assert.deepEqual(state.metadataOffsets, [7], 'the metadata RPC starts at the snapshot offset');
assert.equal(spendOf(owner) - ownerBefore, 110, 'the owner is billed exactly one record');
assert.equal(spendOf(other) - otherBefore, 0, 'the retry account is not billed');
assert.equal(snap().billed_detached, 1);
assert.equal(snap().entries, 0);
});
});
it('SOAK-7 account removal and drain evict entries; the retries reach upstream', async () => {
const owner = seedAccount('soak7-owner');
const other = seedAccount('soak7-other');
createdIds.push(owner.id, other.id);
const { state, handler } = cascadeUpstream();
await withFakeLanguageServer(handler, async (port) => {
const lsFake = makeLsFake(port);
const body = bodyFor('soak7');
await runToCompletion(body, handlerContext(port, owner, lsFake), { closeAfterWrites: 2 });
assert.equal(snap().admit, 1);
auth.removeAccount(owner.id);
assert.ok(await waitFor(() => snap().entries === 0), 'account removal evicts the entry');
assert.equal(snap().evict.account_removed, 1);
assert.equal(lsFake.activeRequests, 0);
state.status = 'idle';
state.grown = true;
const starts0 = state.startCalls;
await runToCompletion(body, handlerContext(port, other, lsFake), {});
assert.equal(state.startCalls - starts0, 1, 'the removed-owner retry goes upstream');
const body2 = bodyFor('soak7-drain');
state.grown = false;
state.status = 'active';
state.firstResumeRead = true;
await runToCompletion(body2, handlerContext(port, other, lsFake), { closeAfterWrites: 2 });
assert.equal(snap().admit, 2);
registry.detachedRegistry.beginDrain();
assert.equal(snap().entries, 0);
assert.equal(snap().draining, true);
assert.equal(lsFake.activeRequests, 0);
const starts1 = state.startCalls;
state.status = 'idle';
state.grown = true;
await runToCompletion(body2, handlerContext(port, other, lsFake), {});
assert.equal(state.startCalls - starts1, 1, 'the drained retry goes upstream');
assert.equal(snap().bytes, 0);
registry.detachedRegistry.reset();
assert.equal(snap().draining, false);
});
});
});