// 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';
import { createRequestBudget } from '../src/request-budget.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 = '3000';
process.env.CASCADE_COLD_STALL_BASE_MS = '3000';
process.env.CASCADE_WARM_STALL_MS = '3000';
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,
    metadataDelayMs: 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));
      };
      // A delayed reply can outlive the caller's budget: the producer bounds
      // the trajectory read by the remaining request time, so it cancels the
      // RPC and Node rejects respond() on the closed stream. That is the
      // expected shape of the deadline tests, not a fake-server bug.
      const respondLater = (delay, body) => {
        const timer = setTimeout(() => {
          if (stream.destroyed || stream.closed) return;
          try { respond(body); } catch { /* stream already cancelled */ }
        }, delay);
        // A cancelled read must not keep the test process alive for the full
        // delay; an awaited metadata read keeps the loop alive on its own.
        timer.unref?.();
      };

      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) {
          respondLater(delay, bodyFor());
        } 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.
        const body = metadataResponse(offset === 0
          ? [metadataEntry(100, 10), metadataEntry(100, 10), metadataEntry(100, 10)]
          : [metadataEntry(100, 10)]);
        if (state.metadataDelayMs > 0) respondLater(state.metadataDelayMs, body);
        else respond(body);
        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,
  };
}

// Two turns so fingerprintBefore() is non-null (reuse needs a prior turn).
function bodyWithHistory(marker) {
  return {
    model: 'gemini-2.5-flash',
    messages: [
      { role: 'user', content: `earlier ${marker}` },
      { role: 'assistant', content: 'earlier answer' },
      { 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,
  });
}

/**
 * Replace the entry under `key` with the same frozen snapshot but a new
 * absolute deadline. Admission arms the registry timer at that deadline;
 * mutating `entry.deadline` afterwards would leave the old timer armed.
 */
function readmitWithDeadline(key, callerKey, frozen, deadlineMs) {
  const current = registry.detachedRegistry.lookup(key, callerKey);
  if (current) {
    assert.equal(registry.detachedRegistry.evict(key, 'resume_miss', current), true,
      'precondition: the current entry is the one being replaced');
  }
  const admitted = registry.detachedRegistry.admit({
    key,
    callerKey,
    snapshot: frozen.snapshot,
    cause: 'client_gone',
    pin: { ...frozen.pin, deadline: Date.now() + deadlineMs, startedAt: Date.now() },
  });
  assert.ok(admitted.token, `re-admission refused: ${admitted.refused}`);
  return admitted;
}

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;
}

// ── T1/T2/T4/T10/T19: admission gates ───────────────────────
describe('Detach P1 admission gates', () => {
  it('T1 an aborted mid-stream turn admits one resumable snapshot', async () => {
    const acct = seedAccount('t1');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t1');
      const ctx = handlerContext(port, acct, lsFake);
      const { state: resState } = await runToCompletion(body, ctx, { closeAfterWrites: 2 });

      assert.equal(resState.closed, true, 'precondition: the client closed mid-stream');
      assert.ok(resState.body.includes(PREFIX), 'precondition: the prefix reached the client');
      assert.equal(registry.detachedRegistry.stats().admit, 1);
      assert.equal(registry.detachedRegistry.stats().refuse.not_client_gone, 0);
      const entry = registry.detachedRegistry.lookup(detachKeyOf(body), 'session:detach-lazy-resume');
      assert.ok(entry, 'the snapshot must be admitted');
      assert.equal(entry.snapshot.version, 2);
      assert.equal(entry.snapshot.consumedThrough, entry.snapshot.stepOffset);
      assert.equal(entry.pin.accountId, acct.id);
      assert.equal(entry.pin.lsGeneration, 1);
      // No lease is pinned beyond the request.
      assert.equal(lsFake.activeRequests, 0, 'a detached entry must not hold beginLsUse');
      assert.equal(state.startCalls, 1);
    });
  });

  it('T2 a pre-content abort admits nothing and the retry reaches upstream', async () => {
    const acct = seedAccount('t2');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t2');
      const ctx = handlerContext(port, acct, lsFake);
      // No producer chunk has arrived yet: the trajectory is still warming,
      // so the response is closed externally. Nothing may be admitted.
      state.stepsFor = () => rawStepsResponse([]);
      const result = await startStream(body, ctx);
      const { res } = makeResponse({});
      const run = result.handler(res);
      await new Promise(resolve => setTimeout(resolve, 25));
      res.emit('close');
      await run;
      assert.equal(registry.detachedRegistry.stats().admit, 0);
      assert.equal(registry.detachedRegistry.stats().refuse.not_client_gone, 0);

      state.stepsFor = null;
      state.status = 'idle';
      state.grown = true;
      await runToCompletion(body, ctx, {});
      assert.equal(state.startCalls, 2, 'a retry with no entry must reach upstream');
    });
  });

  it('T4 a tool-call turn admits nothing and the retry reaches upstream', async () => {
    const acct = seedAccount('t4');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream({
      toolCall: { id: 'call-1', name: 'view_file', argumentsJson: '{"path":"a"}' },
    });
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t4');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 0, 'P1 must not admit a tool turn');

      state.status = 'idle';
      state.grown = true;
      await runToCompletion(body, ctx, {});
      assert.equal(state.startCalls, 2);
    });
  });

  it('T25 a turn that delivered emulated <tool_call> markup admits nothing (D2)', async () => {
    const acct = seedAccount('t25');
    createdIds.push(acct.id);
    const markup = '<tool_call>{"name":"audit_echo","arguments":{"text":"hi"}}</tool_call>';
    const { state, handler } = cascadeUpstream({ initialText: markup });
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = {
        model: 'gemini-2.5-flash',
        messages: [{ role: 'user', content: 'detach test t25' }],
        stream: true,
        tools: [{
          type: 'function',
          function: {
            name: 'audit_echo',
            description: 'echo text',
            parameters: { type: 'object', properties: { text: { type: 'string' } } },
          },
        }],
      };
      const ctx = handlerContext(port, acct, lsFake);
      const { state: resState } = await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.ok(resState.body.includes('audit_echo'),
        'precondition: the emulated call reached the client before the disconnect');
      assert.equal(registry.detachedRegistry.stats().admit, 0,
        'a turn that delivered a tool call must never enter the no-tool detach cache');

      state.status = 'idle';
      state.grown = true;
      await runToCompletion(body, ctx, {});
      assert.equal(state.startCalls, 2,
        'the retry starts a fresh turn instead of replaying the call under a new id');
      assert.equal(state.sendCalls, 2);
    });
  });

  it('T26 a resumed turn that delivers a tool call is not re-frozen (D2)', async () => {
    const acct = seedAccount('t26');
    createdIds.push(acct.id);
    const markup = '<tool_call>{"name":"audit_echo","arguments":{"text":"again"}}</tool_call>';
    const { state, handler } = cascadeUpstream({ initialText: PREFIX });
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = {
        model: 'gemini-2.5-flash',
        messages: [{ role: 'user', content: 'detach test t26' }],
        stream: true,
        tools: [{
          type: 'function',
          function: {
            name: 'audit_echo',
            description: 'echo text',
            parameters: { type: 'object', properties: { text: { type: 'string' } } },
          },
        }],
      };
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1, 'precondition: a tool-free prefix was frozen');

      // The resume grows the SAME step with a complete emulated tool call. The
      // client dies on the call delta: the frozen prefix predates that call, so
      // re-freezing it would replay the call under a new id.
      state.stepsFor = (offset) => rawStepsResponse(offset === 0
        ? [plannerStep(`${PREFIX}${markup}`, { status: 3 })]
        : []);
      const starts0 = state.startCalls;
      const { state: resumeState } = await runToCompletion(body, ctx, { closeAfterWrites: 3 });
      assert.ok(resumeState.body.includes('audit_echo'),
        'precondition: the resumed turn delivered the call before the disconnect');
      assert.equal(registry.detachedRegistry.stats().re_detach, 0,
        'a tool-delivering resume must not be re-frozen');
      assert.equal(registry.detachedRegistry.stats().evict.resume_miss, 1);
      assert.equal(registry.detachedRegistry.stats().entries, 0);
      assert.equal(state.startCalls, starts0, 'the resume itself did not start a cascade');

      // The retry has no entry left, so it goes upstream instead of replaying.
      state.stepsFor = (offset) => rawStepsResponse(offset === 0
        ? [plannerStep(FULL_TEXT, { status: 3 })]
        : []);
      await runToCompletion(body, ctx, {});
      assert.equal(state.startCalls, starts0 + 1, 'the retry starts a fresh turn');
    });
  });

  it('T10 a normal stream end admits nothing', async () => {
    const acct = seedAccount('t10');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t10');
      state.status = 'idle';
      const { state: resState } = await runToCompletion(body, handlerContext(port, acct, lsFake), {});
      assert.equal(resState.writableEnded, true);
      assert.equal(registry.detachedRegistry.stats().admit, 0);
      assert.equal(registry.detachedRegistry.stats().entries, 0);
    });
  });

  it('T19 an idle-empty turn is not stored as complete', async () => {
    const acct = seedAccount('t19');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream({ thinkingOnly: true });
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      state.status = 'idle';
      await runToCompletion(bodyFor('t19'), handlerContext(port, acct, lsFake), {});
      assert.equal(registry.detachedRegistry.stats().admit, 0);
      assert.equal(registry.detachedRegistry.stats().entries, 0);
    });
  });

  it('flag-off parity: with WINDSURFAPI_CASCADE_DETACH unset nothing is admitted', async () => {
    const acct = seedAccount('off');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('off');
      const ctx = handlerContext(port, acct, lsFake);
      delete process.env.WINDSURFAPI_CASCADE_DETACH;
      try {
        await runToCompletion(body, ctx, { closeAfterWrites: 2 });
        assert.equal(registry.detachedRegistry.stats().admit, 0);
        state.status = 'idle';
        state.grown = true;
        await runToCompletion(body, ctx, {});
        assert.equal(state.startCalls, 2, 'flag off keeps the pre-existing retry-to-upstream contract');
      } finally {
        process.env.WINDSURFAPI_CASCADE_DETACH = '1';
      }
    });
  });
});

// ── T5/T7/T11: the resume ───────────────────────────────────
describe('Detach P1 lazy resume', () => {
  it('T5/T7 an identical retry is served from the entry, prefix exactly once, billed to the owner', async () => {
    const owner = seedAccount('t5-owner');
    const other = seedAccount('t5-other');
    createdIds.push(owner.id, other.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t5');
      const ownerCtx = handlerContext(port, owner, lsFake);
      await runToCompletion(body, ownerCtx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);
      assert.equal(spendOf(owner), 0, 'the aborted original is not billed');

      state.status = 'idle';
      state.grown = true;
      const ownerSpendBefore = spendOf(owner);
      const otherSpendBefore = spendOf(other);
      // The retry may be routed to a different account; the resume must not
      // check it out, and must bill the entry owner instead.
      const retryCtx = handlerContext(port, other, lsFake);
      const { state: resState } = await runToCompletion(body, retryCtx, {});

      assert.equal(state.startCalls, 1, 'a resume must not StartCascade again');
      assert.equal(state.sendCalls, 1, 'a resume must not SendUserCascadeMessage again');
      assert.equal(contentOf(resState), FULL_TEXT, 'the answer is the full text, prefix + growth');
      assert.equal(resState.body.split(PREFIX).length - 1, 1, 'the prefix is replayed exactly once');

      const ownerDelta = spendOf(owner) - ownerSpendBefore;
      const otherDelta = spendOf(other) - otherSpendBefore;
      assert.equal(ownerDelta, 330, `billing must land on the owner (got ${ownerDelta})`);
      assert.equal(otherDelta, 0, `the retry account must not be billed (got ${otherDelta})`);
      assert.equal(registry.detachedRegistry.stats().serve, 1);
      assert.equal(registry.detachedRegistry.stats().billed_detached, 1);
      assert.equal(registry.detachedRegistry.stats().entries, 0, 'served entries are deleted');
      assert.equal(lsFake.maxActive, 1, 'exactly one request lease during the resume');
      assert.equal(lsFake.activeRequests, 0, 'the lease is released when the request ends');
      assert.equal(lsFake.acquireCalls, 0, 'leaseHeld must prevent a second acquire');
    });
  });

  it('T3 the metadata RPC starts at the snapshot generatorOffset (no over-bill)', async () => {
    const owner = seedAccount('t3-owner');
    createdIds.push(owner.id);
    const { state, handler } = cascadeUpstream();
    // A resumed pool turn starts at absolute offsets 7/7. The snapshot must
    // carry generatorOffset 7 so the retry's metadata RPC asks for the LAST
    // turn only (1 record), not the whole cascade (3 records).
    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 test t3' },
        ],
        stream: true,
      };
      const ctx = handlerContext(port, owner, lsFake);
      const modelKey = resolveModel(body.model);
      const fpBefore = pool.fingerprintBefore(body.messages, modelKey, ctx.callerKey, {
        tools: [], toolChoice: null, toolPreamble: '', preambleTier: null,
        emulateTools: false, route: 'chat',
      });
      pool.checkin(fpBefore, {
        cascadeId: 'seeded-detach-t3',
        sessionId: 'seeded-session',
        lsPort: port,
        lsGeneration: 1,
        apiKey: owner.apiKey,
        modelKey,
        stepOffset: 7,
        generatorOffset: 7,
      }, ctx.callerKey, 0);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(state.startCalls, 0, 'precondition: the original request resumed the seeded cascade');
      assert.equal(registry.detachedRegistry.stats().admit, 1);
      const entry = registry.detachedRegistry.lookup(detachKeyOf(body), ctx.callerKey);
      assert.equal(entry.snapshot.generatorOffset, 7);
      assert.equal(entry.snapshot.stepOffset, 8);

      state.status = 'idle';
      state.grown = true;
      state.metadataOffsets.length = 0;
      const spendBefore = spendOf(owner);
      await runToCompletion(body, ctx, {});

      assert.deepEqual(state.metadataOffsets, [7],
        `the resume asked metadata from offset(s) ${state.metadataOffsets.join(',')} — 0 re-sums the cascade`);
      assert.equal(spendOf(owner) - spendBefore, 110,
        'the resumed turn must bill only its own record, not all three');
    });
  });

  it('T7b the replayed prefix is sanitized exactly like the live one', async () => {
    const acct = seedAccount('t7b');
    createdIds.push(acct.id);
    const RAW_PATH = '/tmp/windsurf-workspace/secret.js';
    const RAW_PREFIX = `PREFIX-ANSWER-${RAW_PATH} END-`;
    const { state, handler } = cascadeUpstream({ initialText: RAW_PREFIX });
    state.stepsFor = () => rawStepsResponse([
      plannerStep(state.grown ? `${RAW_PREFIX}${GROWTH}` : RAW_PREFIX, { status: state.grown ? 3 : 1 }),
    ]);
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t7b');
      const ctx = handlerContext(port, acct, lsFake);
      const { state: liveState } = await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.ok(liveState.body.includes('<workspace>'), 'precondition: the live prefix was sanitized');
      assert.equal(liveState.body.includes(RAW_PATH), false, 'precondition: the live prefix did not leak the path');
      const entry = registry.detachedRegistry.lookup(detachKeyOf(body), ctx.callerKey);
      assert.ok(entry.snapshot.chunks.some(chunk => String(chunk.text || '').includes(RAW_PATH)),
        'the snapshot stores the raw chunk; the replay must sanitize it');

      state.status = 'idle';
      state.grown = true;
      const { state: resState } = await runToCompletion(body, ctx, {});
      assert.equal(resState.body.includes(RAW_PATH), false, 'the replayed prefix must not leak the raw path');
      assert.ok(contentOf(resState).includes('<workspace>'), 'the replayed prefix carries the sanitizer marker');
      assert.equal(contentOf(resState), `PREFIX-ANSWER-<workspace> END-${GROWTH}`,
        'the resumed answer is the sanitized prefix plus the growth');
      assert.equal(resState.body.split('PREFIX-ANSWER-').length - 1, 1, 'prefix exactly once');
    });
  });

  it('T11 a detached entry pins no LS lease; the resume holds exactly one', async () => {
    const acct = seedAccount('t11');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t11');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(lsFake.activeRequests, 0, 'the detached entry must not keep a beginLsUse pin');

      state.status = 'idle';
      state.grown = true;
      const { state: resState } = await runToCompletion(body, ctx, {});
      assert.equal(lsFake.maxActive, 1, `activeRequests peaked at ${lsFake.maxActive}; a double acquire shows 2`);
      assert.equal(lsFake.activeRequests, 0);
      assert.equal(lsFake.acquireCalls, 0);
      assert.ok(contentOf(resState).includes(GROWTH));
    });
  });
});

// ── T6/T12/T13/T15/T16/T17: negative + positive control ─────
describe('Detach P1 fallbacks', () => {
  it('T6 a different caller scope never sees the entry', async () => {
    const acct = seedAccount('t6');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t6');
      await runToCompletion(body, handlerContext(port, acct, lsFake), { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);

      state.status = 'idle';
      state.grown = true;
      const otherCtx = { ...handlerContext(port, acct, lsFake), callerKey: 'session:someone-else' };
      const { state: resState } = await runToCompletion(body, otherCtx, {});
      assert.equal(state.startCalls, 2, 'a cross-tenant retry must reach upstream');
      assert.equal(contentOf(resState), FULL_TEXT);
      assert.ok(registry.detachedRegistry.stats().lookup.miss >= 1);
    });
  });

  it('T12 an LS generation change evicts the entry and falls back upstream', async () => {
    const acct = seedAccount('t12');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t12');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);

      lsFake.generation = 2; // LS restarted while detached
      state.status = 'idle';
      state.grown = true;
      await runToCompletion(body, ctx, {});
      assert.equal(state.startCalls, 2);
      assert.equal(registry.detachedRegistry.stats().evict.resume_miss, 1);
      assert.equal(registry.detachedRegistry.stats().entries, 0);
      assert.equal(lsFake.activeRequests, 0);
    });
  });

  it('T13 a dead cascade writes zero bytes before falling back upstream', async () => {
    const acct = seedAccount('t13');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t13');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });

      state.failNextSteps = 1; // the resume's first read is not_found
      state.status = 'idle';
      state.grown = true;
      const { state: resState } = await runToCompletion(body, ctx, {});
      assert.equal(state.startCalls, 2, 'the fallback starts a fresh cascade');
      assert.equal(registry.detachedRegistry.stats().evict.resume_miss, 1);
      assert.equal(registry.detachedRegistry.stats().serve, 0);
      // The failed resume wrote nothing; the fresh turn wrote the answer once.
      assert.equal(resState.body.split(PREFIX).length - 1, 1);
      assert.equal(contentOf(resState), FULL_TEXT);
    });
  });

  it('T15 an expired entry is gone and the retry reaches upstream', async () => {
    const acct = seedAccount('t15');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t15');
      const ctx = handlerContext(port, acct, lsFake);
      registry.detachedRegistry.ttlMs = 5;
      try {
        await runToCompletion(body, ctx, { closeAfterWrites: 2 });
        assert.equal(registry.detachedRegistry.stats().admit, 1);
        await new Promise(resolve => setTimeout(resolve, 15));
        registry.detachedRegistry.sweepExpired();
        assert.equal(registry.detachedRegistry.stats().entries, 0);
        assert.equal(registry.detachedRegistry.stats().evict.ttl, 1);
        assert.equal(lsFake.activeRequests, 0, 'expiry releases nothing because no lease was pinned');

        state.status = 'idle';
        state.grown = true;
        await runToCompletion(body, ctx, {});
        assert.equal(state.startCalls, 2);
      } finally {
        registry.detachedRegistry.ttlMs = 120_000;
      }
    });
  });

  it('T16 the per-caller cap evicts the oldest running entry; the evicted retry goes upstream', async () => {
    const acct = seedAccount('t16');
    createdIds.push(acct.id);
    const first = cascadeUpstream();
    const second = cascadeUpstream();
    await withFakeLanguageServer(first.handler, async (port) => {
      // One fake server can serve both turns; swap the handler per phase.
      const lsFake = makeLsFake(port);
      const ctx = handlerContext(port, acct, lsFake);
      const bodyA = bodyFor('t16-a');
      const bodyB = bodyFor('t16-b');
      registry.detachedRegistry.maxPerCaller = 1;
      try {
        await runToCompletion(bodyA, ctx, { closeAfterWrites: 2 });
        assert.equal(registry.detachedRegistry.stats().admit, 1);
        await runToCompletion(bodyB, ctx, { closeAfterWrites: 2 });
        assert.equal(registry.detachedRegistry.stats().admit, 2);
        assert.equal(registry.detachedRegistry.stats().evict.capacity, 1);
        assert.equal(registry.detachedRegistry.lookup(detachKeyOf(bodyA), ctx.callerKey), null);

        first.state.status = 'idle';
        first.state.grown = true;
        const startBefore = first.state.startCalls;
        await runToCompletion(bodyA, ctx, {});
        assert.equal(first.state.startCalls, startBefore + 1, 'the evicted turn goes upstream');
      } finally {
        registry.detachedRegistry.maxPerCaller = 2;
      }
      void second;
    });
  });

  it('T17 removing the owner account evicts the entry and releases nothing else', async () => {
    const owner = seedAccount('t17-owner');
    const other = seedAccount('t17-other');
    createdIds.push(owner.id, other.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t17');
      await runToCompletion(body, handlerContext(port, owner, lsFake), { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);

      auth.removeAccount(owner.id);
      // removeAccount evicts through a dynamic import.
      assert.ok(await waitFor(() => registry.detachedRegistry.stats().entries === 0), 'the entry must be evicted');
      assert.equal(registry.detachedRegistry.stats().evict.account_removed, 1);
      assert.equal(lsFake.activeRequests, 0);

      state.status = 'idle';
      state.grown = true;
      await runToCompletion(body, handlerContext(port, other, lsFake), {});
      assert.equal(state.startCalls, 2);
    });
  });
});

// ── T8/T9: shutdown and drain ───────────────────────────────
describe('Detach P1 server-side causes', () => {
  it('T8 a server shutdown admits nothing', async () => {
    const acct = seedAccount('t8');
    createdIds.push(acct.id);
    const { handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t8');
      const result = await startStream(body, handlerContext(port, acct, lsFake));
      const { res, state } = makeResponse({});
      const run = result.handler(res);
      assert.ok(await waitFor(() => state.body.includes(PREFIX)), 'precondition: the turn streamed');
      const aborted = sse.abortActiveSse('server shutting down');
      assert.ok(aborted >= 1);
      await run;
      assert.equal(registry.detachedRegistry.stats().admit, 0);
      assert.ok(registry.detachedRegistry.stats().refuse.not_client_gone >= 1);
      assert.equal(registry.detachedRegistry.stats().entries, 0);
      assert.equal(lsFake.activeRequests, 0);
    });
  });

  it('T9 a drain timeout admits nothing', async () => {
    const acct = seedAccount('t9');
    createdIds.push(acct.id);
    const { handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t9');
      process.env.WINDSURFAPI_MAX_DRAIN_MS = '1';
      try {
        await runToCompletion(body, handlerContext(port, acct, lsFake), { backpressure: true });
      } finally {
        delete process.env.WINDSURFAPI_MAX_DRAIN_MS;
      }
      assert.equal(registry.detachedRegistry.stats().admit, 0);
      assert.ok(registry.detachedRegistry.stats().refuse.not_client_gone >= 1);
      assert.equal(lsFake.activeRequests, 0);
    });
  });
});

// ── T14/T18: concurrency and re-detach ──────────────────────
describe('Detach P1 claim discipline', () => {
  it('T14 two identical retries: one claims, the other stays foreground', async () => {
    const acct = seedAccount('t14');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t14');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);

      state.status = 'idle';
      state.grown = true;
      state.resumeDelayMs = 60;
      const first = await startStream(body, ctx);
      const firstRes = makeResponse({});
      const firstRun = first.handler(firstRes.res);
      // Give the first request time to claim before the second looks up.
      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(registry.detachedRegistry.stats().claim.ok, 1);
      assert.equal(registry.detachedRegistry.stats().claim.busy, 1);
      assert.equal(registry.detachedRegistry.stats().serve, 1);
      assert.equal(state.startCalls, 2, 'the busy retry starts its own upstream turn');
      assert.ok(contentOf(firstRes.state).includes(GROWTH));
      assert.equal(contentOf(secondRes.state), FULL_TEXT, 'the busy retry still answers from upstream');
      assert.equal(lsFake.activeRequests, 0);
    });
  });

  it('T18 a re-detach keeps the entry claimable and does not extend its TTL', async () => {
    const acct = seedAccount('t18');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t18');
      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;
      // The first retry dies on the FIRST NEW content write: role (1),
      // replayed prefix (2), new growth (3).
      await runToCompletion(body, ctx, { closeAfterWrites: 3 });
      assert.equal(registry.detachedRegistry.stats().re_detach, 1);
      assert.equal(registry.detachedRegistry.stats().serve, 0);
      const reDetached = registry.detachedRegistry.lookup(key, ctx.callerKey);
      assert.ok(reDetached, 'the entry is running again');
      assert.equal(reDetached.expiresAt, originalExpiry, 'a re-detach must not extend the TTL');
      assert.equal(reDetached.snapshot.chunks.map(chunk => chunk.text || '').join(''), FULL_TEXT);

      // The second retry finishes; the final body has each byte exactly once.
      state.firstResumeRead = true;
      const { state: finalState } = await runToCompletion(body, ctx, {});
      assert.equal(registry.detachedRegistry.stats().serve, 1);
      assert.equal(registry.detachedRegistry.stats().entries, 0);
      assert.equal(contentOf(finalState), FULL_TEXT);
      assert.equal(finalState.body.split(PREFIX).length - 1, 1);
      assert.equal(finalState.body.split(GROWTH).length - 1, 1);
      assert.equal(state.startCalls, 1, 'no retry ever started a new cascade');
      assert.equal(state.sendCalls, 1, 'no retry ever re-sent the user message');
    });
  });

  it('T24 a TTL sweep never deletes a live claim; its release still counts as served (D1)', async () => {
    const acct = seedAccount('t24');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const bodyA = bodyFor('t24-a');
      const bodyB = bodyFor('t24-b');
      const ctx = handlerContext(port, acct, lsFake);
      const realNow = registry.detachedRegistry.now;
      const realTtl = registry.detachedRegistry.ttlMs;
      let fakeNow = Date.now();
      registry.detachedRegistry.now = () => fakeNow;
      registry.detachedRegistry.ttlMs = 1_000;
      try {
        // A detaches and a resumer claims it at t=999.
        await runToCompletion(bodyA, ctx, { closeAfterWrites: 2 });
        const key = detachKeyOf(bodyA);
        const claim = registry.detachedRegistry.claim(key, ctx.callerKey);
        assert.ok(claim.entry, 'precondition: the entry is claimable');

        // t=1001: the admission window is gone, but the claim is owned by the
        // original request deadline/token. Another request's admission runs
        // the sweep. Old code deleted the claim here; the owner's release then
        // returned false (serve=0 while the turn was still billed).
        fakeNow += 1_001;
        await runToCompletion(bodyB, ctx, { closeAfterWrites: 2 });
        assert.equal(registry.detachedRegistry.stats().evict.ttl, 0,
          'TTL is an admission bound: it must not evict a claimed entry');
        assert.equal(registry.detachedRegistry.lookup(key, ctx.callerKey)?.state, 'claimed');

        assert.equal(registry.detachedRegistry.release(key, claim.token, { state: 'served' }), true,
          'the original token still owns the entry past its admission window');
        assert.equal(registry.detachedRegistry.stats().serve, 1);
        assert.equal(registry.detachedRegistry.stats().entries, 1, 'only the other request remains');
      } finally {
        registry.detachedRegistry.now = realNow;
        registry.detachedRegistry.ttlMs = realTtl;
      }
    });
  });

  it('T20 a resume that loses its deadline evicts and releases the request lease', async () => {
    const acct = seedAccount('t20');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t20');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      const deadline0 = registry.detachedRegistry.stats().evict.deadline;
      // The claim succeeds, then the request's own budget is already spent:
      // the handler must evict and release the lease it just acquired.
      const budget = {
        deadline: Date.now() + 60_000,
        remainingMs: () => 0,
        waitMs: () => 0,
        takeTurn: () => false,
        expired: () => true,
        describe: () => 'zero-budget',
      };
      const retryCtx = handlerContext(port, acct, lsFake, { budget });
      await runToCompletion(body, retryCtx, {});
      assert.equal(registry.detachedRegistry.stats().evict.deadline - deadline0, 1,
        'the expired resume evicted the entry');
      assert.equal(registry.detachedRegistry.stats().entries, 0);
      assert.equal(registry.detachedRegistry.stats().serve, 0);
      assert.equal(lsFake.activeRequests, 0, 'the claim lease must be released on the deadline path');
      void state;
    });
  });
});

// ── T27–T29: claim deadline races (GPT review F1) ───────────
// The registry timer deletes a claim at the original request deadline. The
// producer must observe that loss: a read bounded by the remaining budget, an
// ownership check after the await and at the replay boundary, and a terminal
// commit gated on release() returning true. Billing still follows the entry's
// stable account id even when the claim died mid-turn.
describe('Detach P1 claim deadline races', () => {
  it('claim_deadline_during_delayed_first_trajectory_read_never_commits_success_without_ownership', async () => {
    const acct = seedAccount('t27');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t27');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1, 'precondition: a snapshot was frozen');

      const key = detachKeyOf(body);
      const frozen = registry.detachedRegistry.lookup(key, ctx.callerKey);
      assert.ok(frozen);
      // 120 ms is all the claim has; the first resume read takes 450 ms.
      readmitWithDeadline(key, ctx.callerKey, frozen, 120);

      state.status = 'idle';
      state.grown = true;
      state.firstResumeRead = true;
      state.resumeDelayMs = 450;

      const deadline0 = registry.detachedRegistry.stats().evict.deadline;
      const serve0 = registry.detachedRegistry.stats().serve;
      const billed0 = registry.detachedRegistry.stats().billed_detached;
      const starts0 = state.startCalls;
      // No fallback turn: the only possible answer is the claim, and its
      // deadline ends while the trajectory read is in flight.
      const budget = createRequestBudget({ deadlineMs: 5_000, upstreamTurnCap: 0 });
      const { state: resState } = await runToCompletion(
        body, handlerContext(port, acct, lsFake, { budget }), {},
      );

      assert.equal(registry.detachedRegistry.stats().serve - serve0, 0,
        'a claim past its deadline must never be served');
      assert.equal(registry.detachedRegistry.stats().evict.deadline - deadline0, 1,
        'the deadline ends the claim exactly once');
      assert.equal(registry.detachedRegistry.stats().entries, 0);
      assert.equal(state.startCalls, starts0,
        'a lost claim must not be replaced by a fresh upstream turn in this request');
      assert.ok(!resState.body.includes('[DONE]'),
        'no success sentinel may be written for a claim the registry no longer owns');
      assert.ok(!contentOf(resState).includes(GROWTH),
        'the delayed read must not replay growth after the deadline');
      assert.equal(resState.writableEnded, true);
      assert.equal(lsFake.activeRequests, 0, 'the request lease is released');
      // The read was cancelled by the remaining budget, so there is no server
      // usage to settle here; the completed-turn settlement is pinned by the
      // metadata-race test below.
      assert.equal(registry.detachedRegistry.stats().billed_detached - billed0, 0);

      // Positive control: the same snapshot with a live deadline is served,
      // and the 330-token settlement still lands on the owner.
      state.firstResumeRead = true;
      state.resumeDelayMs = 0;
      readmitWithDeadline(key, ctx.callerKey, frozen, 60_000);
      const spend0 = spendOf(acct);
      const { state: servedState } = await runToCompletion(body, handlerContext(port, acct, lsFake), {});
      assert.equal(registry.detachedRegistry.stats().serve - serve0, 1,
        'the in-time control is served');
      assert.equal(contentOf(servedState), FULL_TEXT);
      assert.equal(spendOf(acct) - spend0, 330,
        'the in-time resume still bills the owner by stable identity');
    });
  });

  it('claim_evicted_during_first_trajectory_read_never_replays', async () => {
    const acct = seedAccount('t29');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t29');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      const key = detachKeyOf(body);
      assert.ok(registry.detachedRegistry.lookup(key, ctx.callerKey));

      state.status = 'idle';
      state.grown = true;
      state.firstResumeRead = true;
      state.resumeDelayMs = 200;

      const serve0 = registry.detachedRegistry.stats().serve;
      const removed0 = registry.detachedRegistry.stats().evict.account_removed;
      const starts0 = state.startCalls;
      const budget = createRequestBudget({ deadlineMs: 5_000, upstreamTurnCap: 0 });
      const run = runToCompletion(body, handlerContext(port, acct, lsFake, { budget }), {});
      // Evict the claim while the read is in flight. The post-await ownership
      // check must stop the producer before the prefix replay.
      const evicted = await waitFor(() => {
        if (registry.detachedRegistry.lookup(key, ctx.callerKey)?.state !== 'claimed') return false;
        return registry.detachedRegistry.evictForAccount(acct.id) === 1;
      }, 1000);
      assert.equal(evicted, true, 'precondition: the claim was evicted during the read');
      const { state: resState } = await run;

      assert.equal(registry.detachedRegistry.stats().serve - serve0, 0);
      assert.equal(registry.detachedRegistry.stats().evict.account_removed - removed0, 1);
      assert.equal(state.startCalls, starts0, 'no fresh upstream turn (turn cap 0)');
      assert.ok(!resState.body.includes(PREFIX),
        'not one byte of the frozen prefix may be replayed without ownership');
      assert.ok(!resState.body.includes('[DONE]'));
      assert.equal(lsFake.activeRequests, 0);
    });
  });

  it('claim_deadline_after_last_read_still_settles_usage_without_committing_success', async () => {
    const acct = seedAccount('t28');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t28');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      const key = detachKeyOf(body);
      const frozen = registry.detachedRegistry.lookup(key, ctx.callerKey);
      assert.ok(frozen);
      readmitWithDeadline(key, ctx.callerKey, frozen, 120);

      state.status = 'idle';
      state.grown = true;
      state.firstResumeRead = true;
      state.resumeDelayMs = 0;
      // The trajectory reads finish inside the deadline; the post-loop usage
      // RPC outlives it. The claim dies after the turn completed upstream.
      state.metadataDelayMs = 300;

      const deadline0 = registry.detachedRegistry.stats().evict.deadline;
      const serve0 = registry.detachedRegistry.stats().serve;
      const billed0 = registry.detachedRegistry.stats().billed_detached;
      const spend0 = spendOf(acct);
      const budget = createRequestBudget({ deadlineMs: 5_000, upstreamTurnCap: 0 });
      const { state: resState } = await runToCompletion(
        body, handlerContext(port, acct, lsFake, { budget }), {},
      );

      assert.equal(registry.detachedRegistry.stats().serve - serve0, 0,
        'the deadline ended the claim before the success commit');
      assert.equal(registry.detachedRegistry.stats().evict.deadline - deadline0, 1);
      assert.equal(registry.detachedRegistry.stats().billed_detached - billed0, 1,
        'the completed upstream turn is still settled');
      assert.equal(spendOf(acct) - spend0, 330,
        'the 330-token settlement must not be dropped by the ownership gate');
      assert.ok(contentOf(resState).includes(GROWTH),
        'precondition: the turn streamed before its claim ended');
      assert.ok(!resState.body.includes('[DONE]'),
        'no success sentinel after the claim deadline');
      assert.equal(registry.detachedRegistry.stats().entries, 0);
      assert.equal(lsFake.activeRequests, 0);
    });
  });
});

// ── T21–T23: Review360 P2 regressions ───────────────────────
describe('Detach P1 claim/lease regressions', () => {
  it('T21 a synchronous throw after the claim releases the lease and ends the response', async () => {
    const acct = seedAccount('t21');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t21');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);

      state.status = 'idle';
      state.grown = true;
      const begin0 = lsFake.beginCalls;
      const end0 = lsFake.endCalls;
      // The claim succeeds, then the producer cannot even be constructed. The
      // handler used to reject with the lease still held and the socket open.
      class ThrowingClient {
        constructor() { throw new Error('injected constructor failure'); }
      }
      const retryCtx = handlerContext(port, acct, lsFake, { WindsurfClient: ThrowingClient });
      const result = await startStream(body, retryCtx);
      const { res, state: resState } = makeResponse({});
      await assert.rejects(() => result.handler(res), /injected constructor failure/);

      assert.equal(lsFake.beginCalls - begin0, 1, 'the resume acquired exactly one lease');
      assert.equal(lsFake.endCalls - end0, 1, 'the synchronous throw must release it');
      assert.equal(lsFake.activeRequests, 0, 'no request lease may survive the throw');
      assert.equal(resState.writableEnded, true, 'the failed handler must end the response');
      const entry = registry.detachedRegistry.lookup(detachKeyOf(body), ctx.callerKey);
      assert.ok(entry, 'the claim is released back to running, not lost');
      assert.equal(entry.state, 'running');
      assert.equal(registry.detachedRegistry.stats().entries, 1);
    });
  });

  it('T22 a resume miss drops the checked-out cascade instead of resuming it under a new account', async () => {
    const producer = seedAccount('t22-producer');
    const retry = seedAccount('t22-retry');
    createdIds.push(producer.id, retry.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyWithHistory('t22');
      const callerKey = 'session:detach-lazy-resume';
      const accountFor = (a) => ({ id: a.id, email: a.email, apiKey: a.apiKey, proxy: null });

      const producerCtx = handlerContext(port, producer, lsFake, {
        waitForAccount: () => accountFor(producer),
      });
      await runToCompletion(body, producerCtx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);
      assert.equal(state.startCalls, 1);

      // A checked-out pool entry under the retry's fingerprint, owned by the
      // producer account. Only attempt 0 uses its key; the fallback attempt
      // checks out a different account.
      const modelKey = resolveModel(body.model);
      const fp = pool.fingerprintBefore(body.messages, modelKey, callerKey, {
        tools: [], toolChoice: null, toolPreamble: '', preambleTier: null,
        emulateTools: false, route: 'chat',
      });
      assert.ok(fp, 'the two-turn body must have a reuse fingerprint');
      pool.checkin(fp, {
        cascadeId: 'pooled-cascade-t22',
        sessionId: 'pooled-session-t22',
        lsPort: port,
        lsGeneration: lsFake.generation,
        apiKey: producer.apiKey,
        modelKey,
        stepOffset: 1,
        generatorOffset: 0,
      }, callerKey, 0);

      state.status = 'idle';
      state.grown = true;
      state.failNextSteps = 1; // the resume's first read: cascade gone
      let retryAccountCalls = 0;
      const retryCtx = handlerContext(port, retry, lsFake, {
        waitForAccount: () => {
          retryAccountCalls++;
          return accountFor(retry);
        },
      });
      const { state: resState } = await runToCompletion(body, retryCtx, {});

      assert.equal(registry.detachedRegistry.stats().evict.resume_miss, 1);
      assert.ok(retryAccountCalls >= 1, 'the fallback checked out a new account');
      assert.equal(state.startCalls, 2,
        'the fallback must start a fresh cascade; reusing the pool entry would resume it under the retry account');
      assert.equal(state.sendCalls, 2);
      assert.equal(contentOf(resState), FULL_TEXT);
    });
  });

  it('T23 a second retry never evicts an entry another request has claimed', async () => {
    const acct = seedAccount('t23');
    createdIds.push(acct.id);
    const { state, handler } = cascadeUpstream();
    await withFakeLanguageServer(handler, async (port) => {
      const lsFake = makeLsFake(port);
      const body = bodyFor('t23');
      const ctx = handlerContext(port, acct, lsFake);
      await runToCompletion(body, ctx, { closeAfterWrites: 2 });
      assert.equal(registry.detachedRegistry.stats().admit, 1);

      const key = detachKeyOf(body);
      const claim = registry.detachedRegistry.claim(key, ctx.callerKey);
      assert.ok(claim.entry, 'precondition: a live resumer holds the claim');
      const snapshot = claim.entry.snapshot;

      // The claimed resumer's LS restarted while the entry is still claimed.
      // The second retry must stay foreground, not delete the entry.
      lsFake.generation = 2;
      state.status = 'idle';
      state.grown = true;
      const begin0 = lsFake.beginCalls;
      const starts0 = state.startCalls;
      const { state: genState } = await runToCompletion(body, ctx, {});
      assert.equal(registry.detachedRegistry.stats().evict.resume_miss, 0,
        'a claimed entry is never evicted by a second retry');
      assert.equal(registry.detachedRegistry.stats().entries, 1);
      assert.equal(registry.detachedRegistry.lookup(key, ctx.callerKey).state, 'claimed');
      assert.equal(lsFake.beginCalls, begin0, 'a claimed entry is not leased by a second request');
      assert.equal(state.startCalls, starts0 + 1, 'the second retry stays foreground');
      assert.equal(contentOf(genState), FULL_TEXT);

      // A claim past its own request deadline is no longer owned: the next
      // registry operation cleans it by that deadline — never by the second
      // retry's TTL or resume-miss path. Disable timers so the lookup below is
      // the deterministic cleanup point.
      const realTimers = registry.detachedRegistry.timers;
      registry.detachedRegistry.timers = false;
      try {
        const bodyB = bodyFor('t23-deadline');
        const keyB = detachKeyOf(bodyB);
        const admitted = registry.detachedRegistry.admit({
          key: keyB,
          callerKey: ctx.callerKey,
          snapshot,
          cause: 'client_gone',
          pin: {
            cascadeId: 'deadline-cascade', sessionId: 'deadline-session', lsPort: port,
            lsGeneration: lsFake.generation, accountId: acct.id, modelKey: resolveModel(bodyB.model),
            generatorOffset: 0, deadline: Date.now() + 30, startedAt: Date.now(),
          },
        });
        assert.ok(admitted.token);
        const claimB = registry.detachedRegistry.claim(keyB, ctx.callerKey);
        assert.ok(claimB.entry);
        await new Promise(resolve => setTimeout(resolve, 40));
        const ttl0 = registry.detachedRegistry.stats().evict.ttl;
        const miss0 = registry.detachedRegistry.stats().evict.resume_miss;
        const deadline0 = registry.detachedRegistry.stats().evict.deadline;
        const starts1 = state.startCalls;
        const { state: deadlineState } = await runToCompletion(bodyB, ctx, {});
        assert.equal(registry.detachedRegistry.stats().evict.ttl, ttl0,
          'the retry must not evict the claim on TTL');
        assert.equal(registry.detachedRegistry.stats().evict.resume_miss, miss0,
          'the retry must not evict the claim as a resume miss');
        assert.equal(registry.detachedRegistry.stats().evict.deadline - deadline0, 1,
          'the claim is cleaned by its own deadline');
        assert.equal(registry.detachedRegistry.lookup(keyB, ctx.callerKey), null);
        assert.equal(state.startCalls, starts1 + 1);
        assert.equal(contentOf(deadlineState), FULL_TEXT);
        assert.equal(registry.detachedRegistry.release(keyB, claimB.token, { state: 'served' }), false,
          'the deadline already ended the claim');
      } finally {
        registry.detachedRegistry.timers = realTimers;
      }

      registry.detachedRegistry.release(key, claim.token, { state: 'served' });
    });
  });
});