// Three verified defects in src/client.js, pinned at their seams:
//
// 1. The Cascade history budget compared UTF-16 `line.length` while the wire
//    encoder writes UTF-8 (proto.js uses Buffer.byteLength), so a CJK history
//    could "fit" 600 KB and reach the wire at ~1.8 MB.
// 2. The trajectory poll loop never advanced `stepOffset`, so every 500 ms
//    poll re-fetched the whole trajectory (and the resume entry recorded an
//    offset that was really the last batch size).
// 3. `isCascadeTransportError` matched /panel state/i, including the
//    deterministic terminal error the Send retry loop synthesizes after
//    MAX_PANEL_RETRIES — that one rotated an oversized payload across every
//    account instead of surfacing.
//
// The budget tests drive `cascadeChat` against a fake HTTP/2 language server
// and measure the prompt the client actually sends. The offset test uses the
// same seam and asserts on the GetCascadeTrajectorySteps requests. The
// classification test calls the exported pure function.

import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
import http2 from 'node:http2';

Object.assign(process.env, {
  GRPC_PROTOCOL: 'connect',
  CASCADE_POLL_INTERVAL_MS: '10',
  CASCADE_IDLE_GRACE_MS: '1',
  CASCADE_MAX_WAIT_MS: '2000',
  CASCADE_COLD_STALL_BASE_MS: '2000',
  CASCADE_WARM_STALL_MS: '2000',
  WINDSURFAPI_USER_JWT: '0',
});
delete process.env.DEVIN_CONNECT;
delete process.env.DEVIN_ONLY;

const { isCascadeTransportError, WindsurfClient, __TEST_CASCADE_TIMEOUTS } =
  await import('../src/client.js');
const { unwrapRequest, wrapEnvelope, endOfStreamEnvelope } = await import('../src/connect.js');
const {
  getField, parseFields, writeMessageField, writeStringField, writeVarintField,
} = await import('../src/proto.js');

// 3 UTF-8 bytes, 1 UTF-16 code unit.
const CJK = '漢';

function plannerStep(text, status = 3) {
  return Buffer.concat([
    writeVarintField(1, 15), // CORTEX_STEP_TYPE_PLANNER_RESPONSE
    writeVarintField(4, status),
    writeMessageField(20, writeStringField(1, text)),
  ]);
}

function runCommandStep(command, status = 1) {
  return Buffer.concat([
    writeVarintField(1, 28), // CORTEX_STEP_TYPE_RUN_COMMAND
    writeVarintField(4, status),
    writeMessageField(28, writeStringField(23, command)),
  ]);
}

function rawStepsResponse(steps) {
  return Buffer.concat(steps.map((step) => writeMessageField(1, step)));
}

function stepsResponse(texts) {
  return rawStepsResponse(texts.map((text) => plannerStep(text)));
}

function readStepOffset(payload) {
  const field = getField(parseFields(payload), 2, 0);
  return field ? Number(field.value) : 0;
}

function readSendText(payload) {
  const scope = getField(parseFields(payload), 2, 2);
  return getField(parseFields(scope.value), 1, 2).value.toString('utf8');
}

function respond(stream, headers, body) {
  stream.respond({ ':status': 200, 'content-type': headers['content-type'] || 'application/connect+proto' });
  stream.end(Buffer.concat([wrapEnvelope(body, { compress: false }), endOfStreamEnvelope()]));
}

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', handler);
  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));
  }
}

// Runs one fresh cascade and returns the prompt text of every
// SendUserCascadeMessage the fake LS received.
async function runFreshCascade(messages) {
  const sent = [];
  await withFakeLanguageServer((stream, headers) => {
    const chunks = [];
    stream.on('data', (chunk) => chunks.push(chunk));
    stream.on('end', () => {
      const method = String(headers[':path']).split('/').pop();
      const payload = unwrapRequest(Buffer.concat(chunks), headers);
      let body = Buffer.alloc(0);

      if (method === 'StartCascade') body = writeStringField(1, 'budget-cascade');
      if (method === 'SendUserCascadeMessage') sent.push(readSendText(payload));
      if (method === 'GetCascadeTrajectorySteps' && readStepOffset(payload) === 0) {
        body = stepsResponse(['answer']);
      }
      if (method === 'GetCascadeTrajectory') body = writeVarintField(2, 1); // IDLE
      respond(stream, headers, body);
    });
  }, async (port) => {
    const client = new WindsurfClient('test-api-key', port, 'csrf-token');
    await client.cascadeChat(messages, 0, 'claude-sonnet-4-6');
  });
  assert.equal(sent.length, 1, 'exactly one SendUserCascadeMessage per cascade');
  return sent[0];
}

describe('Cascade history budget counts UTF-8 bytes', () => {
  it('trims a CJK history that fits 600 KB as UTF-16 but is ~3x over as UTF-8', async () => {
    const previous = process.env.CASCADE_MAX_HISTORY_BYTES;
    process.env.CASCADE_MAX_HISTORY_BYTES = '600000';
    try {
      // Three 199,000-char turns: 597,024 UTF-16 units each including markup
      // (597,072 total, under the 600 KB budget) but 597,024 UTF-8 bytes each
      // (~1.79 MB total). Counting units kept all three; counting bytes keeps
      // only the newest.
      const prior = Array.from({ length: 3 }, (_, i) => ({
        role: i % 2 ? 'assistant' : 'user',
        content: `${CJK.repeat(199_000)}TURN_${i}`,
      }));
      const text = await runFreshCascade([...prior, { role: 'user', content: 'latest turn' }]);
      const wire = Buffer.byteLength(text, 'utf8');

      assert.ok(wire < 650_000, `prompt is ${wire} bytes; UTF-16 counting sent ~1.79 MB`);
      assert.ok(text.includes('TURN_2'), 'the newest prior turn survives');
      assert.ok(!text.includes('TURN_0') && !text.includes('TURN_1'), 'the older turns are dropped');
      assert.match(text, /truncation_note.*2 earlier turns were dropped/);
    } finally {
      if (previous === undefined) delete process.env.CASCADE_MAX_HISTORY_BYTES;
      else process.env.CASCADE_MAX_HISTORY_BYTES = previous;
    }
  });

  it('counts the prepended system prompt in the same UTF-8 units', async () => {
    const previous = process.env.CASCADE_MAX_HISTORY_BYTES;
    process.env.CASCADE_MAX_HISTORY_BYTES = '600000';
    try {
      // 100,000 CJK chars of system prompt are 300,000 wire bytes (but
      // 100,000 units). Each history line is ~180,024 bytes. Only one line
      // fits after the prompt; counting units would have kept both.
      const system = CJK.repeat(100_000);
      const messages = [
        { role: 'system', content: system },
        { role: 'user', content: `${CJK.repeat(60_000)}TURN_0` },
        { role: 'assistant', content: `${CJK.repeat(60_000)}TURN_1` },
        { role: 'user', content: 'latest turn' },
      ];
      const text = await runFreshCascade(messages);
      const wire = Buffer.byteLength(text, 'utf8');

      assert.ok(wire < 600_000, `prompt is ${wire} bytes; UTF-16 counting sent ~660 KB`);
      assert.ok(text.includes('TURN_1'), 'the newest prior turn survives');
      assert.ok(!text.includes('TURN_0'), 'the older turn no longer fits after the 300 KB prompt');
    } finally {
      if (previous === undefined) delete process.env.CASCADE_MAX_HISTORY_BYTES;
      else process.env.CASCADE_MAX_HISTORY_BYTES = previous;
    }
  });
});

describe('Cascade trajectory polling advances stepOffset with bounded overlap', () => {
  it('advances the poll window while re-reading at most one bounded overlap', async () => {
    const requests = [];
    let produced = 0;

    await withFakeLanguageServer((stream, headers) => {
      const chunks = [];
      stream.on('data', (chunk) => chunks.push(chunk));
      stream.on('end', () => {
        const method = String(headers[':path']).split('/').pop();
        const payload = unwrapRequest(Buffer.concat(chunks), headers);
        let body = Buffer.alloc(0);

        if (method === 'StartCascade') body = writeStringField(1, 'advance-cascade');
        if (method === 'GetCascadeTrajectorySteps') {
          const offset = readStepOffset(payload);
          if (produced < 20) produced += 2;
          const trajectory = Array.from({ length: produced }, (_, i) => plannerStep(`S${i}`));
          requests.push({ offset, count: trajectory.length - offset });
          body = rawStepsResponse(trajectory.slice(offset));
        }
        if (method === 'GetCascadeTrajectory') {
          body = writeVarintField(2, produced >= 20 ? 1 : 2);
        }
        respond(stream, headers, body);
      });
    }, async (port) => {
      const client = new WindsurfClient('test-api-key', port, 'csrf-token');
      const chunks = await client.cascadeChat(
        [{ role: 'user', content: 'go' }], 0, 'claude-sonnet-4-6',
      );
      const text = chunks.map((chunk) => chunk.text || '').join('');
      const offsets = requests.map((request) => request.offset);

      assert.equal(offsets[0], 0);
      assert.deepEqual(offsets, [...offsets].sort((a, b) => a - b), 'the poll window never moves backwards');
      assert.ok(offsets.at(-1) > 0, `the window never advanced past 0: ${offsets.join(',')}`);
      // Source constant CASCADE_STEP_OVERLAP is 8; each request adds at most
      // two new steps, so no response may exceed 10 steps.
      const largest = Math.max(...requests.map((request) => request.count));
      assert.ok(largest <= 10, `a poll re-read ${largest} steps instead of a bounded overlap`);
      assert.equal(text, Array.from({ length: 20 }, (_, i) => `S${i}`).join(''),
        'every step is emitted once, in order, despite the overlap');
      assert.equal(chunks.stepOffset, 20, 'the resume offset is the absolute next step');
    });
  });

  it('re-reads a partial step so growth after the first batch is not lost', async () => {
    let stepPolls = 0;
    let trajectory = [];

    await withFakeLanguageServer((stream, headers) => {
      const chunks = [];
      stream.on('data', (chunk) => chunks.push(chunk));
      stream.on('end', () => {
        const method = String(headers[':path']).split('/').pop();
        const payload = unwrapRequest(Buffer.concat(chunks), headers);
        let body = Buffer.alloc(0);

        if (method === 'StartCascade') body = writeStringField(1, 'growing-cascade');
        if (method === 'GetCascadeTrajectorySteps') {
          stepPolls++;
          if (stepPolls === 1) trajectory = [plannerStep('A', 1)];
          else if (stepPolls === 2) trajectory = [plannerStep('AB', 1), plannerStep('C')];
          else trajectory = [plannerStep('AB', 1), plannerStep('CD')];
          body = rawStepsResponse(trajectory.slice(readStepOffset(payload)));
        }
        if (method === 'GetCascadeTrajectory') {
          body = writeVarintField(2, stepPolls >= 2 ? 1 : 2);
        }
        respond(stream, headers, body);
      });
    }, async (port) => {
      const client = new WindsurfClient('test-api-key', port, 'csrf-token');
      const chunks = await client.cascadeChat(
        [{ role: 'user', content: 'go' }], 0, 'claude-sonnet-4-6',
      );
      const text = chunks.map((chunk) => chunk.text || '').join('');

      // Step 0 arrived as "A", grew to "AB" after the offset advanced, and
      // the IDLE sweep added the final "D" of step 1.
      assert.equal(text, 'ABCD', `partial-step growth was lost: ${JSON.stringify(text)}`);
      assert.equal(chunks.stepOffset, 2, 'both steps are accounted for exactly once');
    });
  });

  it('keeps the tool-active stall ceiling after the tool step leaves the poll window', async () => {
    const savedTimeouts = { ...__TEST_CASCADE_TIMEOUTS };
    Object.assign(__TEST_CASCADE_TIMEOUTS, {
      maxWaitMs: 150,
      pollIntervalMs: 10,
      idleGraceMs: 1,
      coldStallBaseMs: 1000,
      warmStallMs: 50,
      warmStallThinkingMs: 50,
      warmStallToolActiveMs: 300,
    });
    const savedGrace = process.env.CASCADE_TOOL_ACTIVE_GRACE_MS;
    process.env.CASCADE_TOOL_ACTIVE_GRACE_MS = '50';
    try {
      // The ACTIVE tool step is step 0; eight later steps push it out of the
      // bounded overlap window while the tool is still running.
      const trajectory = [
        runCommandStep('git clone https://example.invalid/big.git'),
        ...Array.from({ length: 9 }, (_, i) => plannerStep(`seed${i}`)),
      ];

      await withFakeLanguageServer((stream, headers) => {
        const chunks = [];
        stream.on('data', (chunk) => chunks.push(chunk));
        stream.on('end', () => {
          const method = String(headers[':path']).split('/').pop();
          const payload = unwrapRequest(Buffer.concat(chunks), headers);
          let body = Buffer.alloc(0);

          if (method === 'StartCascade') body = writeStringField(1, 'tool-stall-cascade');
          if (method === 'GetCascadeTrajectorySteps') {
            body = rawStepsResponse(trajectory.slice(readStepOffset(payload)));
          }
          if (method === 'GetCascadeTrajectory') body = writeVarintField(2, 2); // ACTIVE
          respond(stream, headers, body);
        });
      }, async (port) => {
        const client = new WindsurfClient('test-api-key', port, 'csrf-token');
        const started = Date.now();
        const chunks = await client.cascadeChat(
          [{ role: 'user', content: 'run it' }], 0, 'claude-sonnet-4-6',
        );
        const elapsed = Date.now() - started;

        assert.ok(elapsed >= 100, `turn was stall-killed after ${elapsed}ms by the 50ms text ceiling`);
        assert.match(chunks.map((chunk) => chunk.text || '').join(''), /seed0/);
        assert.equal(chunks.toolCalls.length, 1, 'the active tool call is still accounted for');
      });
    } finally {
      Object.assign(__TEST_CASCADE_TIMEOUTS, savedTimeouts);
      if (savedGrace === undefined) delete process.env.CASCADE_TOOL_ACTIVE_GRACE_MS;
      else process.env.CASCADE_TOOL_ACTIVE_GRACE_MS = savedGrace;
    }
  });

  it('keeps resume semantics: starts at the stored offset and returns the absolute next step', async () => {
    const offsets = [];

    await withFakeLanguageServer((stream, headers) => {
      const chunks = [];
      stream.on('data', (chunk) => chunks.push(chunk));
      stream.on('end', () => {
        const method = String(headers[':path']).split('/').pop();
        const payload = unwrapRequest(Buffer.concat(chunks), headers);
        let body = Buffer.alloc(0);

        if (method === 'GetCascadeTrajectorySteps') {
          const offset = readStepOffset(payload);
          offsets.push(offset);
          if (offset === 5) body = stepsResponse(['R5', 'R6']);
        }
        if (method === 'GetCascadeTrajectory') body = writeVarintField(2, 1); // IDLE
        respond(stream, headers, body);
      });
    }, async (port) => {
      const client = new WindsurfClient('test-api-key', port, 'csrf-token');
      const chunks = await client.cascadeChat(
        [{ role: 'user', content: 'continue' }], 0, 'claude-sonnet-4-6',
        {
          reuseEntry: {
            cascadeId: 'resume-cascade',
            sessionId: 'resume-session',
            lsPort: port,
            apiKey: 'test-api-key',
            stepOffset: 5,
            generatorOffset: 5,
          },
        },
      );
      const text = chunks.map((chunk) => chunk.text || '').join('');

      assert.equal(offsets[0], 5, 'resume must poll from the stored offset');
      assert.equal(text, 'R5R6', 'only the new steps are streamed');
      assert.equal(chunks.stepOffset, 7, 'the stored offset plus the new steps');
    });
  });
});

describe('Cascade transport classification', () => {
  it('does not treat the synthesized terminal panel-loss error as transient', () => {
    const terminal = new Error(
      'Panel state lost 2 times after re-warm — likely an LS-level issue with very large '
      + 'payloads (350000 chars). Try reducing system prompt size or tool count.',
    );
    assert.equal(isCascadeTransportError(terminal), false);
  });

  it('still treats genuine panel-state transport errors as transient', () => {
    assert.equal(isCascadeTransportError(new Error('panel state not found for session abc')), true);
    assert.equal(
      isCascadeTransportError(new Error('InitializeCascadePanelState failed: panel state lost')),
      true,
      'an LS-originated panel-state loss (no synthesized "after re-warm" suffix) stays transient',
    );
  });
});