test/devin-connect-priming-recovery.test.js
Rama master · réplica de lectura del repositorio original.
import { afterEach, describe, it } from 'node:test';
import assert from 'node:assert/strict';
import {
streamChatCompletion,
__setStreamChatForTest,
} from '../src/devin-connect-openai.js';
// Regression coverage for the deferred role-priming fix: the role-priming chunk
// must NOT be sent before the upstream demonstrably opens, otherwise the very
// first send() flips `emitted=true` in handlers/chat.js and disarms the
// !emitted-gated account recovery branches (re-login / failover), and it would
// also make the adapter's own pre-emit transport retry look unsafe.
// See src/devin-connect-openai.js streamChatCompletion §1.
afterEach(() => {
__setStreamChatForTest(null);
delete process.env.DEVIN_CONNECT_EAGER_PRIME;
});
function fakeStream(events) {
return async function* () {
for (const ev of events) yield ev;
};
}
function collectSend() {
const frames = [];
return { send: (d) => frames.push(d), frames };
}
const isRole = (f) => f.choices[0]?.delta?.role === 'assistant';
describe('streamChatCompletion deferred priming (recovery safety)', () => {
it('does NOT prime when the FIRST upstream event is a transient error (recovery stays armed)', async () => {
// Mirror the caller: `emitted` flips true on the first send(). If the prime
// fired before the throw, emitted would be true and recovery disarmed.
let emitted = false;
const send = () => { emitted = true; };
__setStreamChatForTest(async function* () {
throw Object.assign(new Error('reset'), { code: 'ECONNRESET' });
// eslint-disable-next-line no-unreachable
yield;
});
await assert.rejects(
streamChatCompletion({ model: 'm', messages: [] }, send, { id: 'x', created: 1 }),
/reset/,
);
assert.equal(emitted, false, 'no byte on the wire → caller can still recover');
});
it('propagates a pre-open error without sending ANY frame', async () => {
const { send, frames } = collectSend();
__setStreamChatForTest(async function* () {
throw Object.assign(new Error('unauthorized'), { code: 'UNAUTHORIZED' });
// eslint-disable-next-line no-unreachable
yield;
});
await assert.rejects(
streamChatCompletion({ model: 'm', messages: [] }, send, { id: 'x', created: 1 }),
/unauthorized/,
);
assert.equal(frames.length, 0, 'zero frames emitted before upstream opened');
});
it('a normal stream produces the SAME event sequence as before (role → reasoning → content → finish → usage)', async () => {
__setStreamChatForTest(fakeStream([
{ type: 'reasoning', text: 'let me think ' },
{ type: 'reasoning', text: 'about it.' },
{ type: 'content', text: 'The answer ' },
{ type: 'content', text: 'is 42.' },
{ type: 'finish', reason: 'stop', usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 } },
]));
const { send, frames } = collectSend();
const result = await streamChatCompletion({ model: 'm', messages: [] }, send, { id: 'x', created: 1, includeUsage: true });
// role chunk is first, and identical to today's shape.
assert.deepEqual(frames[0].choices[0].delta, { role: 'assistant', content: '' });
assert.equal(frames.filter(isRole).length, 1, 'exactly one role chunk');
const kinds = frames.map((f) => {
const d = f.choices[0]?.delta;
if (isRole(f)) return 'role';
if (d?.reasoning_content != null) return 'reasoning';
if (f.choices.length === 0) return 'usage';
if (f.choices[0]?.finish_reason) return 'finish';
if (d?.content != null) return 'content';
return '?';
});
assert.deepEqual(kinds, ['role', 'reasoning', 'reasoning', 'content', 'content', 'finish', 'usage']);
const finish = frames.find((f) => f.choices[0]?.finish_reason === 'stop');
assert.deepEqual(finish.choices[0].delta, {});
assert.equal(result.content, 'The answer is 42.');
assert.equal(result.reasoning, 'let me think about it.');
assert.equal(result.finish_reason, 'stop');
});
it('role chunk is deferred until AFTER the first delta arrives (prime is lazy)', async () => {
// The first frame emitted is the role chunk, but crucially it is only sent
// once the generator has produced its first real event — proven by the
// send happening at all (a pre-open throw would send nothing, see above).
__setStreamChatForTest(fakeStream([
{ type: 'content', text: 'hi' },
{ type: 'finish', reason: 'stop', usage: null },
]));
const { send, frames } = collectSend();
await streamChatCompletion({ model: 'm', messages: [] }, send);
assert.ok(isRole(frames[0]), 'role chunk leads');
assert.equal(frames[1].choices[0].delta.content, 'hi', 'then the content delta');
});
it('an empty / immediate-finish stream still yields a valid finished stream (role → finish)', async () => {
__setStreamChatForTest(fakeStream([
{ type: 'finish', reason: 'stop', usage: null },
]));
const { send, frames } = collectSend();
const result = await streamChatCompletion({ model: 'm', messages: [] }, send, { id: 'x', created: 1 });
// Even with no delta, the client gets a well-formed stream: role then finish.
assert.equal(frames.length, 2);
assert.deepEqual(frames[0].choices[0].delta, { role: 'assistant', content: '' });
assert.deepEqual(frames[1].choices[0].delta, {});
assert.equal(frames[1].choices[0].finish_reason, 'stop');
assert.equal(result.content, '');
assert.equal(result.finish_reason, 'stop');
});
it('an empty stream with usage opt-in still emits the trailing usage frame', async () => {
__setStreamChatForTest(fakeStream([
{ type: 'finish', reason: 'stop', usage: { prompt_tokens: 3, completion_tokens: 0, total_tokens: 3 } },
]));
const { send, frames } = collectSend();
await streamChatCompletion({ model: 'm', messages: [] }, send, { includeUsage: true });
assert.equal(frames.length, 3); // role, finish, usage
assert.deepEqual(frames[0].choices[0].delta, { role: 'assistant', content: '' });
assert.equal(frames[1].choices[0].finish_reason, 'stop');
assert.deepEqual(frames[2].choices, []);
assert.deepEqual(frames[2].usage, { prompt_tokens: 3, completion_tokens: 0, total_tokens: 3 });
});
it('a mid-stream error AFTER a real delta keeps behavior unchanged (already emitted → propagates)', async () => {
// Once a real delta has streamed, prime has fired and the caller's `emitted`
// is true; a subsequent throw is a genuine mid-stream failure that must NOT
// be silently swallowed here. It propagates for the caller to surface.
let emitted = false;
const send = () => { emitted = true; };
__setStreamChatForTest(async function* () {
yield { type: 'content', text: 'partial' };
throw Object.assign(new Error('mid'), { code: 'ECONNRESET' });
});
await assert.rejects(
streamChatCompletion({ model: 'm', messages: [] }, send),
/mid/,
);
assert.equal(emitted, true, 'a delta already reached the wire before the error');
});
it('DEVIN_CONNECT_EAGER_PRIME=1 restores the legacy eager role chunk', async () => {
process.env.DEVIN_CONNECT_EAGER_PRIME = '1';
let framesBeforeThrow = 0;
const send = () => { framesBeforeThrow++; };
__setStreamChatForTest(async function* () {
throw Object.assign(new Error('reset'), { code: 'ECONNRESET' });
// eslint-disable-next-line no-unreachable
yield;
});
// With the opt-in, the role chunk is sent eagerly BEFORE the upstream opens,
// so the frame lands even though the stream then throws (legacy behavior).
await assert.rejects(
streamChatCompletion({ model: 'm', messages: [] }, send),
/reset/,
);
assert.equal(framesBeforeThrow, 1, 'eager prime sent the role chunk up front');
});
it('a tool_call is treated as a real delta and primes the stream', async () => {
__setStreamChatForTest(fakeStream([
{
type: 'finish', reason: 'stop', usage: null,
toolCalls: [{ id: 'c1', name: 'a', arguments: '{}' }],
},
]));
const { send, frames } = collectSend();
const result = await streamChatCompletion({ model: 'm', messages: [] }, send);
// role chunk precedes the tool_calls delta.
assert.ok(isRole(frames[0]), 'role chunk leads');
const toolFrame = frames.find((f) => f.choices[0]?.delta?.tool_calls);
assert.ok(toolFrame, 'tool_calls delta emitted');
assert.equal(frames.filter(isRole).length, 1, 'exactly one role chunk');
assert.equal(result.finish_reason, 'tool_calls');
});
});