src/cascade-turn.js
Rama master · réplica de lectura del repositorio original.
// Producer-level Cascade turn state (Q10 P1, research/Q10/REPORT.md §5.2).
//
// Why this exists: a Cascade trajectory is append-only while a turn is ACTIVE,
// and a step's responseText/thinking grows IN PLACE across polls. The answer
// does not stop at the last step index — E2b measured 596 → 16,632 chars
// inside the same three steps while the client was detached. A resume keyed on
// `stepOffset` alone therefore reads zero steps and returns an empty answer.
//
// The cursor is the design-critical state:
// - `stepOffset` is the absolute index of the next unread step;
// - `pollFloor` is the absolute offset this turn started from, so a resumed
// cascade never replays pre-turn steps as fresh output;
// - `yieldedByStep` / `thinkingByStep` are absolute-index → emitted-length
// cursors, so re-reading the overlap window emits only real growth;
// - the poll window starts at `max(pollFloor, stepOffset - overlap)`, never
// at `stepOffset`, so growth inside an already-counted step is visible.
//
// `snapshot()` / `fromSnapshot()` are the lazy-resume contract. A snapshot
// without the per-step cursors is rejected: it cannot distinguish "already
// emitted" from "not yet read", which is exactly the E2b bug.
// Cascade steps are append-only while ACTIVE: a step's responseText/thinking
// grows in place across polls. Polling strictly after the last consumed step
// would strand that growth until the final sweep, so each poll re-reads this
// many already-seen steps. Bounded (not the whole trajectory) and floored at
// the turn's resume offset so pre-turn steps are never replayed.
export const CASCADE_STEP_OVERLAP = 8;
// v2 (detach P1) adds generatorOffset, sawActive, usageByStep and
// consumedThrough. v1 snapshots were in-memory only, so the bump is free:
// fromSnapshot refuses a v1 payload instead of silently resuming it without
// the metadata offset (over-bill) or the consumed pointer (duplicate steps).
const SNAPSHOT_VERSION = 2;
// The feature flag is read at call time, not module load, so a test (or a
// future runtime config reload) can flip it without re-importing client.js.
// Default OFF: only the exact string '1' enables the producer path.
export function cascadeTurnEnabled(env = process.env) {
return String(env.WINDSURFAPI_CASCADE_TURN || '') === '1';
}
/**
* A snapshot that cannot be resumed safely. Named so callers can distinguish
* a broken entry (drop it, retry upstream) from a transport failure.
*/
export class CascadeTurnCursorError extends Error {
constructor(message) {
super(message);
this.name = 'CascadeTurnCursorError';
this.code = 'CASCADE_TURN_CURSOR';
}
}
function pairsFromMap(value, label) {
if (value instanceof Map) {
for (const [index, length] of value) {
if (!Number.isInteger(index) || index < 0 || !Number.isInteger(length) || length < 0) {
throw new CascadeTurnCursorError(`CascadeTurn ${label} has a non-integer cursor entry`);
}
}
return new Map(value);
}
if (!Array.isArray(value)) {
throw new CascadeTurnCursorError(`CascadeTurn ${label} must be a Map or an array of [index, length] pairs`);
}
const map = new Map();
for (const pair of value) {
if (!Array.isArray(pair) || pair.length !== 2) {
throw new CascadeTurnCursorError(`CascadeTurn ${label} entry is not an [index, length] pair`);
}
const [index, length] = pair;
if (!Number.isInteger(index) || index < 0 || !Number.isInteger(length) || length < 0) {
throw new CascadeTurnCursorError(`CascadeTurn ${label} has a non-integer cursor entry`);
}
map.set(index, length);
}
return map;
}
function positiveInt(value, fallback) {
return Number.isInteger(value) && value > 0 ? value : fallback;
}
/**
* Per-step usage is an index → {inputTokens, outputTokens, cacheReadTokens,
* cacheWriteTokens} map. JSON-safe form is an array of [index, usage] pairs.
* A non-object value is a broken snapshot, not a usage report: refuse it
* rather than resume with a silently dropped billing fallback.
*/
function usageFromMap(value) {
const entries = value instanceof Map ? [...value.entries()] : value;
if (!Array.isArray(entries)) {
throw new CascadeTurnCursorError('CascadeTurn usageByStep must be a Map or an array of [index, usage] pairs');
}
const map = new Map();
for (const pair of entries) {
if (!Array.isArray(pair) || pair.length !== 2) {
throw new CascadeTurnCursorError('CascadeTurn usageByStep entry is not an [index, usage] pair');
}
const [index, usage] = pair;
if (!Number.isInteger(index) || index < 0) {
throw new CascadeTurnCursorError('CascadeTurn usageByStep has a non-integer index');
}
if (!usage || typeof usage !== 'object') {
throw new CascadeTurnCursorError('CascadeTurn usageByStep has a non-object usage value');
}
map.set(index, { ...usage });
}
return map;
}
export class CascadeTurn {
/**
* @param {object} [options]
* @param {string} [options.cascadeId]
* @param {number} [options.stepOffset=0] absolute index of the next unread step
* @param {number} [options.pollFloor=stepOffset] absolute offset this turn started from
* @param {number} [options.overlap=CASCADE_STEP_OVERLAP] bounded re-read window
* @param {Array} [options.chunks=[]] producer buffer (raw chunks, not SSE bytes)
* @param {(chunk: object) => any} [options.emit] sink called after each chunk is buffered
* @param {Map|Array} [options.yieldedByStep]
* @param {Map|Array} [options.thinkingByStep]
* @param {number} [options.totalYielded=0]
* @param {number} [options.totalThinking=0]
* @param {boolean} [options.sawText=false]
* @param {number} [options.generatorOffset=0] metadata-RPC offset this turn started from
* @param {boolean} [options.sawActive=false] true once the trajectory was non-IDLE
* @param {Map|Array} [options.usageByStep] per-step usage fallback
* @param {number} [options.consumedThrough=pollFloor] max(consumed index)+1
*/
constructor({
cascadeId = '',
stepOffset = 0,
pollFloor = stepOffset,
overlap = CASCADE_STEP_OVERLAP,
chunks = [],
emit = null,
yieldedByStep = new Map(),
thinkingByStep = new Map(),
totalYielded = 0,
totalThinking = 0,
sawText = false,
generatorOffset = 0,
sawActive = false,
usageByStep = new Map(),
consumedThrough = pollFloor,
} = {}) {
if (!Number.isInteger(stepOffset) || stepOffset < 0) {
throw new CascadeTurnCursorError(`CascadeTurn stepOffset must be a non-negative integer, got ${stepOffset}`);
}
if (!Number.isInteger(pollFloor) || pollFloor < 0 || pollFloor > stepOffset) {
throw new CascadeTurnCursorError(`CascadeTurn pollFloor must be an integer in [0, stepOffset], got ${pollFloor}`);
}
if (!Number.isInteger(generatorOffset) || generatorOffset < 0) {
throw new CascadeTurnCursorError(`CascadeTurn generatorOffset must be a non-negative integer, got ${generatorOffset}`);
}
if (!Number.isInteger(consumedThrough) || consumedThrough < 0 || consumedThrough > stepOffset) {
throw new CascadeTurnCursorError(
`CascadeTurn consumedThrough must be an integer in [0, stepOffset], got ${consumedThrough}`,
);
}
this.cascadeId = cascadeId;
this.stepOffset = stepOffset;
this.pollFloor = pollFloor;
this.overlap = positiveInt(overlap, CASCADE_STEP_OVERLAP);
this.chunks = chunks;
this.emit = typeof emit === 'function' ? emit : null;
this.yieldedByStep = pairsFromMap(yieldedByStep, 'yieldedByStep');
this.thinkingByStep = pairsFromMap(thinkingByStep, 'thinkingByStep');
this.totalYielded = Number.isFinite(totalYielded) ? totalYielded : 0;
this.totalThinking = Number.isFinite(totalThinking) ? totalThinking : 0;
this.sawText = !!sawText;
this.generatorOffset = generatorOffset;
this.sawActive = !!sawActive;
this.usageByStep = usageFromMap(usageByStep);
// Absolute pointer proving every counted step was actually consumed. A
// detach is only safe when this equals stepOffset: otherwise the steps
// between the two would fall out of the resume window (stepOffset -
// overlap) and their text would be lost.
this.consumedThrough = consumedThrough;
this.lastWindowStart = pollFloor;
this.lastGrowthAt = Date.now();
// Set by fromSnapshot: the exact window the detached turn would have read
// next. Used once, then the live formula takes over.
this.resumeWindowStart = null;
}
/** Window start the next poll would use. Does not consume a pending resume. */
nextWindowStart() {
if (this.resumeWindowStart !== null) return this.resumeWindowStart;
return Math.max(this.pollFloor, this.stepOffset - this.overlap);
}
/** The produced text so far (all chunks, including pre-detach). */
get text() {
let out = '';
for (const chunk of this.chunks) out += chunk?.text || '';
return out;
}
/** The produced thinking so far. */
get thinking() {
let out = '';
for (const chunk of this.chunks) out += chunk?.thinking || '';
return out;
}
/**
* One GetCascadeTrajectorySteps read. Advances `stepOffset` and records the
* window; does NOT consume the steps (call consumeStep in the caller's order
* so tool-call chunks interleave exactly as they do in the foreground loop).
*
* @param {(windowStart: number) => Promise<Array>} fetchSteps
*/
async readWindow(fetchSteps) {
if (typeof fetchSteps !== 'function') {
throw new TypeError('CascadeTurn.readWindow requires a fetchSteps(windowStart) function');
}
const windowStart = this.nextWindowStart();
const steps = await fetchSteps(windowStart);
if (!Array.isArray(steps)) {
throw new TypeError('CascadeTurn fetchSteps must resolve to an array of steps');
}
// Only consume the pending resume window after a successful read: a
// transient fetch failure must retry the same (older, wider) window.
this.resumeWindowStart = null;
this.lastWindowStart = windowStart;
const previousStepOffset = this.stepOffset;
const nextStepOffset = windowStart + steps.length;
if (nextStepOffset > this.stepOffset) this.stepOffset = nextStepOffset;
const advanced = this.stepOffset > previousStepOffset;
if (advanced) this.lastGrowthAt = Date.now();
return {
windowStart,
steps,
advanced,
stepOffset: this.stepOffset,
};
}
/**
* Emit the thinking/text growth of one step since its cursor. The cursor is
* advanced before the sink await, matching the foreground loop: a slow or
* failed consumer must not cause the same delta to be emitted twice.
*
* @returns {Promise<{textDelta: string, thinkingDelta: string, grew: boolean}>}
*/
async consumeStep(step, stepIndex) {
// The step was read and is being consumed: advance the proof pointer
// BEFORE any await, so a detach landing inside the sink still records it.
// `stepIndex < stepOffset` keeps a standalone consume (no readWindow) from
// moving the pointer past a cursor the turn never advanced.
if (stepIndex < this.stepOffset && stepIndex + 1 > this.consumedThrough) {
this.consumedThrough = stepIndex + 1;
}
let thinkingDelta = '';
const liveThink = step?.thinking || '';
if (liveThink) {
const prevThink = this.thinkingByStep.get(stepIndex) || 0;
if (liveThink.length > prevThink) {
thinkingDelta = liveThink.slice(prevThink);
this.thinkingByStep.set(stepIndex, liveThink.length);
this.totalThinking += thinkingDelta.length;
this.lastGrowthAt = Date.now();
await this.#emit({ text: '', thinking: thinkingDelta, isError: false });
}
}
let textDelta = '';
const liveText = step?.responseText || step?.text || '';
if (liveText) {
const prev = this.yieldedByStep.get(stepIndex) || 0;
if (liveText.length > prev) {
textDelta = liveText.slice(prev);
this.yieldedByStep.set(stepIndex, liveText.length);
this.totalYielded += textDelta.length;
this.sawText = true;
this.lastGrowthAt = Date.now();
await this.#emit({ text: textDelta, thinking: '', isError: false });
}
}
return { textDelta, thinkingDelta, grew: !!(textDelta || thinkingDelta) };
}
/**
* The final IDLE sweep's polish pass. `responseText` is topped up first;
* `modifiedText` is only appended when it is a strict extension of
* responseText (a rewrite would splice wrong bytes onto the stream). Both
* top-ups mark `sawText`: this sweep is often the FIRST place a turn's text
* becomes visible, and a turn that emitted text did not end empty.
*/
async consumeFinalStep(step, stepIndex) {
if (stepIndex < this.stepOffset && stepIndex + 1 > this.consumedThrough) {
this.consumedThrough = stepIndex + 1;
}
const responseText = step?.responseText || '';
const modifiedText = step?.modifiedText || '';
let textDelta = '';
const prev = this.yieldedByStep.get(stepIndex) || 0;
if (responseText.length > prev) {
textDelta = responseText.slice(prev);
this.yieldedByStep.set(stepIndex, responseText.length);
this.totalYielded += textDelta.length;
// F2: the sweep can be the FIRST place this step's text is visible, so
// it must account for the emission exactly like consumeStep does.
// Leaving sawText false made the turn end `idle_empty` after the bytes
// had already been streamed.
this.sawText = true;
this.lastGrowthAt = Date.now();
await this.#emit({ text: textDelta, thinking: '', isError: false });
}
const cursor = this.yieldedByStep.get(stepIndex) || 0;
if (modifiedText.length > cursor && modifiedText.startsWith(responseText)) {
const polish = modifiedText.slice(cursor);
this.yieldedByStep.set(stepIndex, modifiedText.length);
this.totalYielded += polish.length;
this.sawText = true;
this.lastGrowthAt = Date.now();
await this.#emit({ text: polish, thinking: '', isError: false });
textDelta += polish;
}
return { textDelta };
}
/**
* Fetch + consume in one call. This is the pump's entry point; the
* foreground loop uses readWindow/consumeStep separately so its per-step
* tool-call handling keeps its exact ordering.
*
* `beforeStep` runs before the step's deltas are consumed and may return
* false to stop (e.g. the native-tool proposal hand-off). `afterStep` runs
* after the deltas.
*/
async pollOnce(fetchSteps, { beforeStep, afterStep } = {}) {
const read = await this.readWindow(fetchSteps);
let grew = read.advanced;
let stopped = false;
for (let i = 0; i < read.steps.length; i++) {
const step = read.steps[i];
const stepIndex = read.windowStart + i;
if (typeof beforeStep === 'function' && (await beforeStep(step, stepIndex)) === false) {
stopped = true;
break;
}
const consumed = await this.consumeStep(step, stepIndex);
grew = grew || consumed.grew;
await afterStep?.(step, stepIndex, consumed);
}
return { ...read, grew, stopped };
}
/** Move the next-unread cursor forward (the final sweep can see new steps). */
advanceTo(stepOffset) {
if (Number.isInteger(stepOffset) && stepOffset > this.stepOffset) {
this.stepOffset = stepOffset;
}
return this.stepOffset;
}
/**
* The resumable producer state. The cursor maps are JSON-safe [index,
* length] arrays; chunks are shallow copies (nested tool payloads keep
* their original object identity).
*/
snapshot() {
// A detach taken mid-batch would store stepOffset ahead of the cursors;
// steps older than stepOffset - overlap would then fall out of the resume
// window and their text would be lost forever. Refuse instead of storing
// an entry that cannot be resumed correctly.
if (this.consumedThrough !== this.stepOffset) {
throw new CascadeTurnCursorError(
`CascadeTurn cannot snapshot a mid-batch turn: consumedThrough ${this.consumedThrough} != stepOffset ${this.stepOffset}`,
);
}
return {
version: SNAPSHOT_VERSION,
cascadeId: this.cascadeId,
stepOffset: this.stepOffset,
pollFloor: this.pollFloor,
windowStart: this.nextWindowStart(),
lastWindowStart: this.lastWindowStart,
overlap: this.overlap,
yieldedByStep: [...this.yieldedByStep.entries()].sort((a, b) => a[0] - b[0]),
thinkingByStep: [...this.thinkingByStep.entries()].sort((a, b) => a[0] - b[0]),
totalYielded: this.totalYielded,
totalThinking: this.totalThinking,
sawText: this.sawText,
// v2: the metadata RPC is offset-scoped. Resuming from 0 re-sums every
// prior turn of the cascade and over-bills it (src/windsurf.js:947-951).
generatorOffset: this.generatorOffset,
// v2: the IDLE grace only applies before the trajectory was ever
// non-IDLE; a resumed ACTIVE turn must not wait the extra grace.
sawActive: this.sawActive,
// v2: fallback usage when the metadata RPC fails.
usageByStep: [...this.usageByStep.entries()].sort((a, b) => a[0] - b[0]).map(([i, u]) => [i, { ...u }]),
// v2: proves every counted step was consumed (see the guard above).
consumedThrough: this.consumedThrough,
chunks: this.chunks.map(chunk => ({ ...chunk })),
};
}
/**
* Restore a detached turn. Refuses a snapshot without per-step cursors:
* `stepOffset` alone cannot tell which bytes were already emitted, and
* starting the window there is the E2b "0 steps / 0 chars" bug.
*
* @param {object} snapshot
* @param {object} [options]
* @param {(chunk: object) => any} [options.emit]
*/
static fromSnapshot(snapshot, { emit = null } = {}) {
if (!snapshot || typeof snapshot !== 'object') {
throw new CascadeTurnCursorError('CascadeTurn.fromSnapshot requires a snapshot object');
}
if (snapshot.version != null && snapshot.version !== SNAPSHOT_VERSION) {
throw new CascadeTurnCursorError(`CascadeTurn snapshot version ${snapshot.version} is not supported (expected ${SNAPSHOT_VERSION})`);
}
if (!Number.isInteger(snapshot.stepOffset) || snapshot.stepOffset < 0) {
throw new CascadeTurnCursorError('CascadeTurn snapshot has no valid stepOffset');
}
if (!Number.isInteger(snapshot.pollFloor) || snapshot.pollFloor < 0 || snapshot.pollFloor > snapshot.stepOffset) {
throw new CascadeTurnCursorError('CascadeTurn snapshot has no valid pollFloor');
}
if (!Array.isArray(snapshot.yieldedByStep) || !Array.isArray(snapshot.thinkingByStep)) {
throw new CascadeTurnCursorError(
'CascadeTurn snapshot has no per-step cursors (yieldedByStep/thinkingByStep); '
+ 'resuming from stepOffset alone returns zero steps (Q10 E2b: stepOffset=3 → 0 steps / 0 chars)',
);
}
// v2 validity precondition: every counted step must have been consumed.
// Without this an entry with stepOffset=3 and empty cursors is accepted,
// and the resumed window (max(0, 3-8)=0) re-emits steps 0-2 — a duplicated
// prefix, not the E2b "0 steps" bug.
if (!Number.isInteger(snapshot.consumedThrough)) {
throw new CascadeTurnCursorError('CascadeTurn snapshot has no consumedThrough pointer');
}
if (snapshot.consumedThrough !== snapshot.stepOffset) {
throw new CascadeTurnCursorError(
`CascadeTurn snapshot is mid-batch: consumedThrough ${snapshot.consumedThrough} != stepOffset ${snapshot.stepOffset}`,
);
}
if (!Number.isInteger(snapshot.generatorOffset) || snapshot.generatorOffset < 0) {
throw new CascadeTurnCursorError(
'CascadeTurn snapshot has no generatorOffset; resuming the metadata RPC from 0 re-bills prior turns',
);
}
if (typeof snapshot.sawActive !== 'boolean') {
throw new CascadeTurnCursorError('CascadeTurn snapshot has no sawActive flag');
}
if (!Array.isArray(snapshot.usageByStep)) {
throw new CascadeTurnCursorError('CascadeTurn snapshot has no usageByStep fallback');
}
const turn = new CascadeTurn({
cascadeId: snapshot.cascadeId,
stepOffset: snapshot.stepOffset,
pollFloor: snapshot.pollFloor,
overlap: snapshot.overlap,
chunks: Array.isArray(snapshot.chunks)
? snapshot.chunks.map(chunk => (chunk && typeof chunk === 'object' ? { ...chunk } : chunk))
: [],
emit,
yieldedByStep: snapshot.yieldedByStep,
thinkingByStep: snapshot.thinkingByStep,
totalYielded: snapshot.totalYielded,
totalThinking: snapshot.totalThinking,
sawText: snapshot.sawText,
generatorOffset: snapshot.generatorOffset,
sawActive: snapshot.sawActive,
usageByStep: snapshot.usageByStep,
consumedThrough: snapshot.consumedThrough,
});
if (Number.isInteger(snapshot.windowStart)) {
if (snapshot.windowStart < snapshot.pollFloor || snapshot.windowStart > snapshot.stepOffset) {
throw new CascadeTurnCursorError(
`CascadeTurn snapshot windowStart ${snapshot.windowStart} is outside [${snapshot.pollFloor}, ${snapshot.stepOffset}]`,
);
}
turn.resumeWindowStart = snapshot.windowStart;
}
if (Number.isInteger(snapshot.lastWindowStart)) turn.lastWindowStart = snapshot.lastWindowStart;
return turn;
}
async #emit(chunk) {
this.chunks.push(chunk);
if (this.emit) await this.emit(chunk);
}
}