Verify the tee interception on a live stream

Spike proves response.body.tee() splits a real routerai SSE stream: the
consumer sees chunks progressively (first at 1ms of 1643ms) and our branch
still recovers the generation id and usage.cost. It also found a real
defect -- a rebuilt Response loses url, which the OpenAI SDK reads -- and
that Object.defineProperty is the only way to restore it.
This commit is contained in:
2026-10-15 13:02:32 +03:00
parent 82f8b7baef
commit c9d2314bb2
@@ -80,10 +80,24 @@ this.fetch = options.fetch ?? Shims.getDefaultFetch(); // openai/client.js:160
```js ```js
const [дляНас, потребителю] = response.body.tee(); const [дляНас, потребителю] = response.body.tee();
наблюдать(дляНас); // разбираем SSE/JSON в фоне наблюдать(дляНас); // разбираем SSE/JSON в фоне
return new Response(потребителю, { status, statusText, headers: response.headers }); const rebuilt = new Response(потребителю, { status, statusText, headers: response.headers });
Object.defineProperty(rebuilt, 'url', { value: response.url, configurable: true });
return rebuilt;
``` ```
`tee()` — штатный примитив разделения потока: обе ветви текут инкрементально, ничего не буферизуется целиком, харнесс получает байты без задержки. Из своей ветви обёртка достаёт `usage.cost` и `id`; для не-потоковых ответов разбирает JSON так же. **Это проверено на живом потоке routerai** (`.recon/spike-tee.mjs`), а не выведено умозрительно:
- потребитель получил 9041 байт за 19 чтений, первый чанк через **1 мс**, последний через 1643 мс — поток течёт инкрементально, задержки нет;
- наша ветвь увидела 30 SSE-чанков, достала `id` генерации и `cost = 0.01082233360065`;
- `content-type: text/event-stream`, `content-encoding` и `content-length` пусты, поэтому пересборка заголовков на потоке ничего не портит.
**Дефект, который нашёлся и уже закрыт:** у пересобранного `Response` поле `url` пустое, а OpenAI SDK его читает (`response.url` в логах повторов, `openai/client.js:432,446,458`). Присваивание напрямую падает:
```
Cannot assign to read only property 'url' of object '#<_Response>'
```
Работает только `Object.defineProperty` — проверено в `.recon/spike-url.mjs`. Без этой строки SDK, скорее всего, не сломался бы (поле используется только в отладочном выводе), но полагаться на это не стоит.
Результат кладётся в реестр **ключом «id генерации → стоимость»**. Результат кладётся в реестр **ключом «id генерации → стоимость»**.
@@ -169,7 +183,7 @@ return new Response(потребителю, { status, statusText, headers: respo
**3. `tee()` и backpressure.** Обе ветви надо читать. Свою ветвь мы читаем сразу и до конца; если этого не делать, буфер потребителя будет расти. Требование к реализации: свою ветвь никогда не оставлять невычитанной. **3. `tee()` и backpressure.** Обе ветви надо читать. Свою ветвь мы читаем сразу и до конца; если этого не делать, буфер потребителя будет расти. Требование к реализации: свою ветвь никогда не оставлять невычитанной.
**4. Конструирование нового `Response`.** `new Response(potок, {status, headers})` теряет поле `url` исходного ответа. Надо проверить, не полагается ли на него SDK; если полагается — переносить вручную. **4. Конструирование нового `Response`.** ~~`new Response(potок, {status, headers})` теряет поле `url` исходного ответа.~~ **Проверено и закрыто:** `url` действительно теряется, SDK его читает, поэтому поле восстанавливается через `Object.defineProperty`. Прямое присваивание не работает (read-only), подробности и команда проверки — в разделе про обёртку `fetch`.
## Технические решения ## Технические решения
@@ -187,7 +201,6 @@ return new Response(потребителю, { status, statusText, headers: respo
- Порядок загрузки, гарантирующий, что обёртка встанет до первого вызова модели. - Порядок загрузки, гарантирующий, что обёртка встанет до первого вызова модели.
- Точный отбор запросов по URL и разбор SSE без потери чанков на границах буфера. - Точный отбор запросов по URL и разбор SSE без потери чанков на границах буфера.
- Сохранение `url` (или подтверждение, что оно не нужно) при конструировании нового `Response`.
- Структура реестров и где их хранить; как переживать перезапуск. - Структура реестров и где их хранить; как переживать перезапуск.
- Разрешение гонки «стоимость и пара пришли в разном порядке»; политика для так и не собранных пар. - Разрешение гонки «стоимость и пара пришли в разном порядке»; политика для так и не собранных пар.
- Выбор между опросом и push для доставки числа. - Выбор между опросом и push для доставки числа.
@@ -203,6 +216,8 @@ return new Response(потребителю, { status, statusText, headers: respo
| `probe-routerai.mjs` | сырой ответ routerai: заголовки и тело с `usage.cost` | | `probe-routerai.mjs` | сырой ответ routerai: заголовки и тело с `usage.cost` |
| `probe-stream.mjs` | то же для потокового ответа | | `probe-stream.mjs` | то же для потокового ответа |
| `probe-immediate.mjs` | доступность `generation?id=…` сразу после ответа | | `probe-immediate.mjs` | доступность `generation?id=…` сразу после ответа |
| `spike-tee.mjs` | **спайк механизма:** `tee()` на живом потоке — инкрементальность, извлечение `cost`, потеря `url` |
| `spike-url.mjs` | восстановление `url` на пересобранном `Response` |
| `probe-generation.mjs` | `generation?id=…` и цены endpoint'ов с `variable_pricings` | | `probe-generation.mjs` | `generation?id=…` и цены endpoint'ов с `variable_pricings` |
| `audit-usage.mjs` | что лежит в durable-логах сессий | | `audit-usage.mjs` | что лежит в durable-логах сессий |
| `asar.mjs`, `list.mjs`, `extract.mjs`, `grep2.mjs` | чтение `app.asar`: список, извлечение, поиск | | `asar.mjs`, `list.mjs`, `extract.mjs`, `grep2.mjs` | чтение `app.asar`: список, извлечение, поиск |