vvzvlad/gitmost
1
0
Fork 0
Code Issues 17 Pull Requests 14 Packages Projects Releases Activity

feat(ai-chat): единый resumable SSE-транспорт для агентских ранов — attach/replay вместо поллинга (#184, фаза 1.5) #381

New Issue
Open
opened 2026-07-06 00:08:36 +03:00 by agent_vscode · 0 comments
agent_vscode commented 2026-07-06 00:08:36 +03:00
Collaborator
Copy Link
Copy Source

Зачем

PR #234 (фаза 1 #184) сделал раны durable: сервер доводит ход после дисконнекта браузера. Но UI-доставка осталась двухпутёвой: вкладка-стартер видит живой SSE-стрим, а переоткрытая вкладка — 2-секундный поллинг снапшотов POST /ai-chat/run без токен-стрима. Вся latch-механика F4/F5/F7 (run-polling.ts, ai-chat-window.tsx) существует только потому, что два транспорта дерутся за вью.

Цель: один транспорт для всех вкладок. Ран публикует свой SSE-стрим в серверный реестр; любая вкладка подключается через GET-attach с реплеем кадров и живым хвостом. Наблюдатель становится обычным стримером (status === 'streaming') → Stop-кнопка, прыгающие точки и инвалидации работают штатными путями. Поллинг и латчи удаляются.

⚠️ Правила для исполнителя — прочитать до начала

  1. Все решения приняты. Не отклоняться, не «улучшать». Невозможно сделать как написано / противоречие — СТОП и вопрос комментарием в ишью.
  2. Два PR строго по порядку: PR 1 (сервер, «спящий»), потом PR 2 (клиент + удаление). Не смешивать.
  3. Комментарии в коде — только на английском. Точечные правки.
  4. Не трогать: legacy-путь (флаг off — байт-в-байт; generateMessageId поэтому гейтится на runId), семантику POST /ai-chat/stream, POST /ai-chat/run, POST /ai-chat/stop, public-share чат, схему БД (миграций нет).
  5. Не добавлять зависимостей (всё есть в установленном AI SDK: server ai@6.0.134 патченный, client ai@6.0.207 + @ai-sdk/react@3.0.209). Никакого resumable-stream/Redis/BullMQ.
  6. Известный смежный баг мейнлайна — НЕ чинить здесь: «Send now» у стартера в autonomous-режиме не делает server-stop → флаш может словить 409. Отдельная задача.

Проверенные факты об SDK (сверены по установленным исходникам — опираться, не перепроверять)

  • pipeUIMessageStreamToResponse(res, { consumeSseStream }) тее-ит поток SSE-строк до сокета (server dist:5911-5920); tee-ветка живёт при смерти клиентского сокета. Кадры data: {json}\n\n, терминальный data: [DONE]\n\n. Heartbeat'ы (: ping) в tee не попадают.
  • reconnectToStream: GET ${api}/${chatId}/stream, chatId = store id useChat (не серверный!), без AbortSignal (client dist:13334-13372). prepareReconnectToStreamRequest полностью заменяет URL. Только 204 → null (тихий no-op, статус не меняется, onFinish НЕ зовётся); другой не-2xx бросает → onError + красный баннер (и onFinish тоже НЕ зовётся — см. задачу 2.3 п. 3).
  • resumeStream() — mount-only эффект (react dist:221-225). Терминал resumed-стрима зовёт onFinish из finally (dist:13757-13766) — как у локального.
  • Сид-стрип обязателен: text-start всегда пушит новую парту (dist:5746-5754) — реплей поверх засеянной in-progress строки дублирует текст. То же верно для реплея ЗАВЕРШЁННОГО рана поверх settled-строки — поэтому settled-хвост никогда не резюмим.
  • Аборт: чанк {"type":"abort"} → [DONE] → конец. Клиентский процессор abort-чанк игнорирует → у вкладки, не нажимавшей Stop, это чистый финиш (isAbort:false). Принято как паритет — но см. задачу 2.3 п. 8 (подавление флаша). Error-чанк, напротив, БРОСАЕТ в процессоре (dist:6109-6111 → 13734-13736) → isError + баннер.
  • Локальный stop() НЕ абортит resumed-фетч (сигнал не проброшен), но взводит isAbort=true на будущий onFinish (dist:13686-13688) — источник «зомби»-ловушек, закрытых ниже.
  • assistantId присваивается awaited-вставкой (ai-chat.service.ts:842-853) ДО pipe-вызова (:1176) на обоих путях. generateMessageId без originalMessages штампует start.messageId (dist:5819-5834). Chat id — UUIDv7, ParseUUIDPipe по умолчанию (all) его принимает.
  • Порядок в AiChatService.stream: beginRun :574-580 → вставка user-строки :609 → сид ассистент-строки :842 (между ними — загрузка истории, настройки, MCP-хендшейки — секунды!) → pipe :1176. На этом порядке держатся open()-at-begin гарантии.
  • Гонка refetch ↔ finalizeAssistant: терминальная запись ассистент-строки — первый await внутри onFinish у streamText (:1026-1039), стартует примерно когда [DONE] уходит на провод. Инвалидация сразу после клиентского onFinish регулярно возвращает строку ещё status:'streaming' (со снапшотом последнего ЗАВЕРШЁННОГО шага, для одношагового ответа — возможно пустым сидом). Поэтому мержить streaming-снапшот поверх полного стримленного сообщения НЕЛЬЗЯ — п. 7 задачи 2.3 взводит recovery только когда мержить действительно нужно.
  • TanStack v5: query.state.fetchFailureCount СБРАСЫВАЕТСЯ в 0 в начале каждого фетча (query-core dist:353-358, 408-412) — межтиковые consecutive-ошибки им НЕ выразить. Поэтому degraded-poll НЕ гасится по ошибкам вообще (см. задачу 2.4): единственные границы — 10-мин кап, settled, локальный стрим, смена чата. Бонус: поллинг переживает рестарт сервера.

ИНВАРИАНТЫ (нарушение любого = сломанный дизайн; на каждый есть тест)

  1. bind/abortEntry мутируют entry только при совпадении runId.
  2. Retention-делит удаляет только свою entry: if (entries.get(chatId) === entry) entries.delete(chatId).
  3. open() при замене завершает старую entry зеркально done-пути (started-подписчикам onEnd() И удалить их; paused — pendingEnd=true; снять её таймер) — включая ещё «живую» (ран settled, tee-петля дочитывает).
  4. Снапшот реплея и регистрация подписчика — в одном синхронном блоке внутри attach (paused-режим); между ними нет await.
  5. В контроллере cleanup (close-хендлер → unsubscribe) регистрируется до первой записи в сокет, плюс явная проверка req.raw.destroyed после регистрации (close мог прийти во время await'ов); реплей+start() — в try/catch с unsubscribe.
  6. finished-entry при expect=live отдаётся БЕЗ регистрации подписчика; без expect=live — 204. При expect=live с anchor несовпадение entry.assistantMessageId !== anchor → 204 (кросс-рановый реплей запрещён).
  7. Resumed-ход никогда не флашит очередь (ни plain-flush, ни flush-on-abort).
  8. Degraded-poll гаснет ТОЛЬКО по четырём причинам: локальный стрим (тред зовёт onResumeFallback(false)), settled-merge (тред зовёт onResumeFallback(false)), смена чата, 10-мин кап. Ошибки fetch поллинг НЕ гасят.
  9. Здоровый финиш resumed-стрима НИЧЕГО не мержит из кэша в стор; restore стрипнутой строки — ТОЛЬКО при отсутствии видимого контента в пришедшем message (клоббер более свежего частичного стрима mount-time снапшотом запрещён — в т.ч. на isDisconnect).

PR 1 — сервер

Задача 1.1. apps/server/src/core/ai-chat/ai-chat-stream-registry.service.ts (новый)

export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000;
export const RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024;
// 2x the replay cap: a just-written 4MB replay burst alone can never trip the
// per-subscriber cap (see controller); only a genuinely stalled socket can.
export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES;

export interface RunStreamCallbacks {
  onFrame: (frame: string) => void;
  onEnd: () => void;
}
export interface RunStreamAttachment {
  replay: string[];
  finished: boolean;
  start(): void;        // drain pending frames (order preserved) and go live
  unsubscribe(): void;  // safe to call at any point, idempotent
}

interface Subscriber extends RunStreamCallbacks {
  started: boolean;
  pending: string[];
  pendingEnd: boolean;
}
interface Entry {
  runId: string;
  // The persisted assistant row id of this run (set at bind; undefined if the
  // seed failed). Used by the attach anchor check (invariant 6).
  assistantMessageId?: string;
  frames: string[];
  bytes: number;
  overflowed: boolean;
  finished: boolean;
  subscribers: Set<Subscriber>;
  retainTimer?: NodeJS.Timeout;
}

@Injectable()
export class AiChatStreamRegistryService implements OnModuleDestroy {
  private readonly entries = new Map<string, Entry>(); // key: chatId

  open(chatId: string, runId: string): void;
  bind(chatId: string, runId: string, assistantMessageId: string | undefined,
       stream: ReadableStream<string>): void;
  abortEntry(chatId: string, runId: string): void;
  attach(chatId: string, expectLive: boolean, anchor: string | undefined,
         cb: RunStreamCallbacks): Promise<RunStreamAttachment | null>; // async: phase-2 Redis seam
  onModuleDestroy(): void; // clear all retain timers
}

Семантика (реализовать ровно так):

  • open (в момент старта рана, до любых кадров): если по chatId есть старая entry — завершить её ЗЕРКАЛЬНО done-пути (инвариант 3: finished=true, started-подписчикам onEnd() и удалить их из Set, paused — pendingEnd=true, снять её retainTimer; поздний done её tee-петли сработает по замкнутой ссылке и не тронет новую благодаря identity-проверкам; повторный onEnd невозможен — started уже удалены), затем entries.set(chatId, { runId, frames: [], bytes: 0, overflowed: false, finished: false, subscribers: new Set() }).
  • bind (из consumeSseStream): найти entry; если нет или entry.runId !== runId — warn-лог и выход (инвариант 1). Иначе записать entry.assistantMessageId = assistantMessageId и fire-and-forget цикл reader.read():
    • кадр: frames.push(value); bytes += Buffer.byteLength(value); затем если bytes > RUN_STREAM_MAX_BUFFER_BYTES → overflowed = true; frames = []; (один warn-лог; пересёкший кадр УЧТЁН и разослан; после overflow пуши в frames прекращаются, рассылка живым продолжается). Рассылка: каждому подписчику — started → try { sub.onFrame(frame) } catch { subscribers.delete(sub) }; paused → sub.pending.push(frame).
    • done ИЛИ ошибка чтения: finished = true; started-подписчикам onEnd() и удалить; paused — pendingEnd = true (их дренирует start()); взвести retention-таймер с identity-проверкой (инвариант 2), .unref().
  • abortEntry (из внешнего catch стрим-метода): identity-проверка по runId; если entry ещё не finished — тот же терминальный путь, что у done.
  • attach — синхронно, без await между шагами: entry нет ИЛИ overflowed → null; expectLive && anchor && entry.assistantMessageId !== anchor → null (инвариант 6: стрипнувшая вкладка не должна получить реплей ЧУЖОГО рана — например, N+1 финишировал и N+2 уже открыт в суб-секундном окне между fetch истории и attach); finished && !expectLive → null; finished && expectLive → { replay: frames.slice(), finished: true, start: noop, unsubscribe: noop } (подписчик НЕ регистрируется); иначе зарегистрировать paused-подписчика (started:false, pending:[], pendingEnd:false) и вернуть { replay: frames.slice(), finished: false, start, unsubscribe }. start(): синхронно доставить pending по порядку через onFrame, затем started = true; если pendingEnd — onEnd() и удалить. unsubscribe(): удалить из Set (идемпотентно).
  • Примечание к anchor: если bind ещё не случился (окно begin→pipe), entry.assistantMessageId undefined → мисматч с любым anchor → 204 → клиентский restore+poll путь сходится сам. Это норма, не чинить.

Задача 1.2. Модуль

ai-chat.module.ts: AiChatStreamRegistryService в providers.

Задача 1.3. Врезки в существующий код

  1. ai-chat.controller.ts, begin-хук (замыкание runHooks ~:345-374): сразу после успешного beginRun (там, где известен runId) добавить this.streamRegistry.open(chatId, runId);. Реестр инжектится в контроллер.
  2. ai-chat.service.ts, pipe-вызов (~:1176), добавить в options ровно два поля — ОБА гейтятся на runId (правило 4: legacy-путь при флаге off — байт-в-байт, включая отсутствие start.messageId; сид assistantId при этом безусловный, поэтому гейт только по assistantId менял бы legacy-провод):
// #184 phase 1.5: run-wrapped mode only — the legacy path (flag off) stays
// byte-for-byte identical, including the absence of start.messageId.
...(runId
  ? {
      // Tee the SSE frames into the run-stream registry so late tabs can
      // attach (replay + live tail).
      consumeSseStream: ({ stream }: { stream: ReadableStream<string> }) =>
        this.streamRegistry.bind(chatId, runId, assistantId, stream),
      // Stamp the persisted assistant row's DB id onto the streamed message so
      // every tab renders the SAME id as the DB row (id-based reconciliation).
      // Seeding is best-effort: when it failed, let the client generate the id.
      ...(assistantId ? { generateMessageId: () => assistantId } : {}),
    }
  : {}),
  1. ai-chat.service.ts, внешний catch стрим-метода (~:1241, рядом с аварийным onSettled): if (runId) this.streamRegistry.abortEntry(chatId, runId); — иначе подписчики открытой entry зависнут навсегда.

Задача 1.4. Attach-эндпоинт

ai-chat.controller.ts, роут GET /api/ai-chat/runs/:chatId/stream (+ query expect, anchor). @UseGuards/@Throttle — та же комбинация, что у stream-эндпоинта, лимит { limit: 60, ttl: 60000 }; @SkipTransform(). (Бакеты throttler'а раздельные по хендлерам — проверено, с 25/мин стрима не пересекается.)

@Get('runs/:chatId/stream')
async attachRunStream(
  @Param('chatId', new ParseUUIDPipe()) chatId: string,
  @Query('expect') expect: string | undefined,
  @Query('anchor') anchor: string | undefined,
  @Req() req: FastifyRequest,
  @Res() res: FastifyReply,
  @AuthUser() user: User,
  @AuthWorkspace() workspace: Workspace,
) {
  await this.assertOwnedChat(chatId, user, workspace); // same gate as getRun
  let stopHeartbeat: () => void = () => undefined;
  const attachment = await this.streamRegistry.attach(
    chatId,
    expect === 'live',
    anchor,
    {
      onFrame: (frame) => {
        // Backpressure guard: 2x the replay cap, so the initial replay burst
        // alone can never trip it; only a genuinely stalled socket can.
        try {
          if (res.raw.writableLength > SUBSCRIBER_MAX_BUFFERED_BYTES) {
            res.raw.destroy(); // 'close' fires -> unsubscribe below
            return;
          }
          if (!res.raw.writableEnded) res.raw.write(frame);
        } catch {
          res.raw.destroy();
        }
      },
      onEnd: () => {
        stopHeartbeat();
        if (!res.raw.writableEnded) res.raw.end();
      },
    },
  );
  if (!attachment) {
    res.status(204).send(); // the ONLY "nothing to resume" signal the SDK accepts
    return;
  }
  res.hijack();
  // Cleanup BEFORE any write (invariant 5): a torn-down socket must not orphan
  // a paused subscriber whose pending queue would buffer the whole run.
  req.raw.once('close', () => {
    attachment.unsubscribe();
    stopHeartbeat();
  });
  // A close emitted DURING the awaits above was missed by the listener — check.
  // (Healthy pending GETs have req.raw.destroyed === false, so no false
  // positives; returning without end() is fine — the socket is gone.)
  if (req.raw.destroyed) {
    attachment.unsubscribe();
    return;
  }
  res.raw.on('error', () => undefined);
  try {
    res.raw.writeHead(200, {
      'content-type': 'text/event-stream',
      'cache-control': 'no-cache',
      'x-vercel-ai-ui-message-stream': 'v1',
      'x-accel-buffering': 'no',
      // deliberately NO Connection/Keep-Alive (hop-by-hop; Safari/HTTP2)
    });
    res.raw.flushHeaders?.();
    for (const frame of attachment.replay) res.raw.write(frame);
    if (attachment.finished) {
      res.raw.end();
      return;
    }
    stopHeartbeat = startSseHeartbeat(res.raw, 15_000);
    attachment.start(); // drain pending accumulated during replay, go live
  } catch {
    attachment.unsubscribe();
    stopHeartbeat();
    res.raw.destroy();
  }
}

Задача 1.5. Тесты PR 1

  • registry.spec (unit): реплей в порядке поступления; поздний attach = полный префикс + живой хвост; paused-подписчик получает pending в порядке, затем live (кадры между attach и start не теряются и не переупорядочиваются); finished во время паузы → onEnd на start(); finished+expect=live → реплей без регистрации подписчика; finished без expect → null; anchor-мисматч при expect=live → null (и до bind — assistantMessageId undefined → null); совпадение → attach; overflow → attach null, живой подписчик продолжает получать (оракул: push→учёт→проверка, пересёкший кадр разослан); open() поверх retained entry завершает старую и снимает её таймер; open() поверх LIVE entry (ран settled, tee ещё читает): started-подписчики получают ровно ОДИН onEnd (удалены при open — поздний done её петли не дублирует вызов), поздний done не удаляет новую entry (identity, инвариант 3); retention удаляет только свою entry; bind/abortEntry с чужим runId — no-op; бросающий onFrame выкидывает только этого подписчика, ingest-цикл жив.
  • controller.attach.spec (по образцу ai-chat.controller.run.spec.ts): 403 чужой чат; 204 (нет entry / finished без expect=live / anchor-мисматч); реплей пишется; close до start() → unsubscribe; close во время await'ов (req.raw.destroyed к моменту регистрации) → unsubscribe, ничего не пишется; cap → destroy; finished-реплей завершает ответ сразу.
  • int-spec (test/integration/ai-chat-attach.int-spec.ts, по образцу stream-int-спека с MockLanguageModelV3): дисконнект стартера на середине → attach → кадры идентичны полной последовательности вкл. [DONE]; attach СРАЗУ после старта рана (до первого кадра) получает поток с нулевого кадра; requestStop → подписчик видит {"type":"abort"} + [DONE] + конец.
  • Расширить ai-chat.service.spec.ts: consumeSseStream И generateMessageId присутствуют ⇔ run-wrapped (легаси-опции pipe-вызова НЕ содержат ни того, ни другого); generateMessageId = id засеянной строки; abortEntry зовётся из внешнего catch.

DoD PR 1: tsc + build; существующие тесты зелёные без правок (кроме ai-chat.service.spec.ts); новые зелёные; флаг off — байт-в-байт (вкл. отсутствие start.messageId). Флаг on со СТАРЫМ клиентом: единственное видимое изменение — start.messageId = DB-id строки; старый клиент к id не привязан (единственный потребитель — mergeObservedMessage, которому совпадение id только помогает); намеренно и допустимо.

PR 2 — клиент

Задача 2.1. Типы

types/ai-chat.types.ts: в IAiChatMessageRow добавить status?: string; (провод уже несёт — baseFields включают status).

Задача 2.2. utils/resume-helpers.ts (новый)

  • isStreamingTail(rows) — хвост role==='assistant' && status==='streaming'.
  • isSettledAssistantTail(rows) — хвост-ассистент со status !== 'streaming'.
  • seedRows(rows, strip) — при strip вернуть без последней строки.
  • mergeById(messages, incoming) — перенос СУЩЕСТВУЮЩЕЙ mergeObservedMessage из run-polling.ts (логика не меняется).
  • Тесты: перенесённые кейсы merge + новые для обоих предикатов и seedRows.

Задача 2.3. components/chat-thread.tsx

Разрешён ровно один новый проп: onResumeFallback?: (active: boolean) => void (управление degraded-poll в окне, задача 2.4).

  1. Объявления (все новые ref'ы/стейт этого блока):
// resume machinery refs (#184 phase 1.5)
const attachAbortRef = useRef<AbortController | null>(null);
const reconcileTailRef = useRef(false);
const noStreamHandledRef = useRef(false);
const onNoActiveStreamRef = useRef<(() => void) | null>(null);
const [resumedTurn, setResumedTurn] = useState(false);
const resumedTurnRef = useRef(false);
// Identity-stable pair setter (bare useState setter + ref write): it is closed
// over by the transport useMemo([]), so it MUST NOT capture state.
const setResumedTurnPair = useCallback((v: boolean) => {
  resumedTurnRef.current = v;
  setResumedTurn(v);
}, []);
  1. Гейтинг (mount-time, в ref'ах):
// Attempt resume for any non-settled tail: a streaming tail (strip + expect
// live replay) or a user tail (the run may exist but its assistant row is not
// seeded yet — attach to the pre-opened registry entry and wait for frames).
// A settled assistant tail must NEVER resume: replaying a finished run into a
// store that already contains its message duplicates parts (SDK text-start
// always pushes a new part).
const stripRef = useRef(chatId !== null && isStreamingTail(initialRows ?? []));
const attemptResumeRef = useRef(
  autonomousRunsEnabled === true && chatId !== null &&
    !isSettledAssistantTail(initialRows ?? []),
);
const strippedRowRef = useRef<IAiChatMessageRow | null>(
  stripRef.current ? (initialRows ?? [])[initialRows!.length - 1] : null,
);
  1. Сид: seedRows(initialRows ?? [], attemptResumeRef.current && stripRef.current).map(rowToUiMessage).
  2. Транспорт — два новых поля в существующем useMemo:
prepareReconnectToStreamRequest: () => ({
  // SDK default URL uses the useChat STORE id — always build from the real chat id.
  // ?expect=live&anchor=<row id> ONLY when we stripped a streaming tail: expect=live
  // is the only case where a finished-retained replay is safe (the row is stripped,
  // replay rebuilds it), and the anchor pins the replay to OUR run — a mismatching
  // (newer) run must 204 into the restore+poll path instead of replaying a foreign
  // transcript into this store.
  api: `/api/ai-chat/runs/${chatIdRef.current}/stream${
    stripRef.current ? `?expect=live&anchor=${strippedRowRef.current!.id}` : ''
  }`,
}),
fetch: async (input: RequestInfo | URL, init: RequestInit = {}) => {
  if ((init.method ?? 'GET') !== 'GET') return fetch(input, init); // send path untouched
  // Reconnect GET: the SDK passes no AbortSignal, so wire our own controller
  // for observer Stop; surface the silent 204; and clear the resumedTurn flag
  // on ANY no-onFinish outcome (204, HTTP error, network throw) so it cannot
  // suppress the NEXT local turn's queue flush.
  const controller = new AbortController();
  attachAbortRef.current = controller;
  try {
    const response = await fetch(input, { ...init, signal: controller.signal });
    if (response.status === 204) onNoActiveStreamRef.current?.();
    else if (!response.ok) setResumedTurnPair(false); // resume error: no onFinish will come
    return response;
  } catch (err) {
    setResumedTurnPair(false); // network failure: no onFinish will come
    throw err;
  }
},
  1. Mount-эффект:
useEffect(() => {
  if (attemptResumeRef.current) {
    setResumedTurnPair(true);
    void resumeStream();
  }
  // Mount-only by design; the parent remounts per chat via `key`.
  // eslint-disable-next-line react-hooks/exhaustive-deps
}, []);
  1. 204-обработчик (onNoActiveStream, one-shot через noStreamHandledRef — НЕ занулять onNoActiveStreamRef), строго четыре части:
    • (а) только если strippedRowRef.current — вернуть строку в стор через mergeById (user-tail 204 сюда НЕ заходит — не разыменовывать null!);
    • (б) безусловно — reconcileTailRef.current = true + инвалидация AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current);
    • (в) безусловно — onResumeFallback?.(true). Поллинг окна — тупой таймер с 10-мин капом (задача 2.4); тред гасит его сам через onResumeFallback(false) на settled/локальном стриме. На user-tail 204 без рана он дотикает до капа впустую (0.4 rps, принято); если ран стартовал в окне send→begin — очередной тик застанет seed и degraded-следование дойдёт до settle;
    • (г) setResumedTurnPair(false) (204 = onFinish не будет; иначе флаг подавил бы флаш СЛЕДУЮЩЕГО локального хода).
  2. Реконсиляция + degraded-мердж — один эффект с deps ровно [initialRows, isStreaming, setMessages]:
    • при isStreaming === true → сбросить reconcileTailRef, вызвать onResumeFallback?.(false) (инвариант 8: локальный ход гасит и мердж, и поллинг), выход;
    • иначе, если reconcileTailRef.current === true и последняя строка initialRows — ассистент: mergeById её в стор НА КАЖДОЕ обновление initialRows (пока degraded-poll активен, это и есть живой пошаговый прогресс); если её status !== 'streaming' (settled) — сбросить reconcileTailRef И вызвать onResumeFallback?.(false) (терминальный merge сделан — гасим поллинг явно; окно само его не погасит, у него только временной кап).
  3. Recovery после голодного/оборванного resumed-финиша (инвариант 9). В onFinish, используя wasResumed из п. 8; арм и restore гейтятся ПО-РАЗНОМУ:
    • арм (reconcile + инвалидация messages + onResumeFallback?.(true)): если wasResumed && (isDisconnect || !assistantMessageHasVisibleContent(message)) (реюз утилиты из utils/message-content.ts);
    • restore стрипнутой строки (mergeById(strippedRowRef.current)): ТОЛЬКО если wasResumed && !assistantMessageHasVisibleContent(message) && strippedRowRef.current. На isDisconnect С видимым контентом restore ЗАПРЕЩЁН: живой стрим мог уйти далеко вперёд mount-time снапшота, и restore откатил бы контент на экране (клоббер, инвариант 9) — там достаточно арма: поллинг доведёт до настоящего терминала.
      Здоровый resumed-финиш НЕ взводит НИЧЕГО: стор уже держит полное стримленное сообщение, обновление кэша придёт через onTurnFinished, а немедленный refetch гонится с finalizeAssistant (см. факт в шапке). Provider-error mid-run: error-чанк бросает в процессоре → isError + баннер; при видимом контенте recovery не нужен, при пустом — условие арма и так истинно.
  4. Порядок в onFinish — строго (инвариант 7): (1) const wasResumed = resumedTurnRef.current; и сразу setResumedTurnPair(false); (2) recovery п. 7 (по wasResumed); (3) стандартные ветки — при wasResumed === true пропустить ОБЕ ветки отправки: plain flushNext() И ветку flushOnAbortRef; onTurnFinished и прочее — как обычно (возможная двойная инвалидация messages после п. 7 — намеренная и безвредная, НЕ «дедуплицировать»). Дополнительно флаг сбрасывается при любом ЛОКАЛЬНОМ send (все три call-site sendMessage: onSend, flushNext, sendNow — перед вызовом setResumedTurnPair(false)).
  5. «Send now» на resumed-ходе скрыт: в JSX очереди кнопка interrupt-send не рендерится при resumedTurn (стейт из п. 0; остаётся только удаление). Причина: локальный stop() не абортит resumed-фетч (зомби: клик «проглочен», а flushOnAbortRef выстрелил бы спустя минуты на естественном финише); вариант «стопнуть сервер и слать» отвергнут из-за 409-гонки с асинхронным finalize.
  6. Stop: в handleStop добавить первой строкой attachAbortRef.current?.abort();. Остальное не менять. (Наблюдатель, нажавший Stop: isAbort с частичным контентом → recovery по п. 7 не взводится, restore не делается — частичное остаётся + нотис «Response stopped.»; персистентный aborted-маркер появится при следующем открытии — паритет со стартером.)

Задача 2.4. components/ai-chat-window.tsx — degraded-poll фолбэк (замена латчей)

  • Стейт const [degradedPoll, setDegradedPoll] = useState(false); + armedAtRef (штамп Date.now() при включении); сбрасывать при смене activeChatId (инвариант 8).
  • Проп onResumeFallback={...} на <ChatThread> (сеттер, штампующий armedAt при true).
  • Константа DEGRADED_POLL_MAX_MS = 10 * 60_000.
  • useAiChatMessagesQuery получает опцию refetchInterval (прокинуть через хук), callback-форма — НАМЕРЕННО ТУПАЯ: degradedPoll === true && Date.now() - armedAtRef.current < DEGRADED_POLL_MAX_MS ? 2500 : false. НИКАКИХ проверок ошибок (fetchFailureCount в TanStack v5 сбрасывается на каждом фетче и consecutive-ошибки не выражает — см. факт в шапке; поллинг обязан переживать рестарт сервера) и НИКАКИХ проверок хвоста (semantics живут в треде: settled/локальный стрим гасят через onResumeFallback(false), п. 6 задачи 2.3). Временной кап — единственная страховка от вечного тика (застрявшая 'streaming'-строка до boot-sweep; user-tail без рана).
  • Доставка контента в degraded-режиме — целиком на стороне ChatThread (задача 2.3 п. 6); окно только владеет таймером.

Задача 2.5. Список удаления (точный)

  • utils/run-polling.ts, utils/run-polling.test.ts — удалить (merge-логика и её кейсы переехали в задаче 2.2).
  • queries/ai-chat-query.ts — удалить useAiChatRunQuery, AI_CHAT_RUN_RQ_KEY; queries/ai-chat-run-query.test.tsx — удалить.
  • services/ai-chat-service.ts — удалить getAiChatRun (⚠️ stopRun оставить).
  • types/ai-chat.types.ts — удалить IAiChatRun, IAiChatRunResponse.
  • components/ai-chat-window.tsx — удалить: localStreaming+onStreamingChange; латч stoppingRun + reset; removeQueries из handleServerStop (сжимается до stopRun(chatId).catch(notify)); блок useAiChatRunQuery; F7-эффект; isObservingRun/observedRow; терминальный эффект (finalizedRunIdRef); пропсы observedRow/onStreamingChange; мёртвые импорты.
  • components/chat-thread.tsx — удалить проп observedRow + observer-merge эффект; проп onStreamingChange + эффект.
  • components/chat-thread.test.tsx — удалить describe «observer run merge (#184)».
  • Серверный POST /ai-chat/run не трогать.

Задача 2.6. Тесты PR 2

chat-thread.test.tsx (мок-харнесс уже ловит транспорт/useChat; добавить спай resumeStream):

  • resumeStream на маунте ⇔ флаг on ∧ chatId ∧ хвост НЕ settled-ассистент; сид стрипается только при streaming-хвосте; user-хвост — attempt без стрипа;
  • URL: ?expect=live&anchor=<id стрипнутой строки> только при стрипе; без стрипа — без query;
  • 204 при user-хвосте: НЕТ краша, НЕТ restore, ЕСТЬ reconcile+invalidate, ЕСТЬ onResumeFallback(true);
  • 204 при streaming-хвосте: restore + invalidate + onResumeFallback(true) + сброс resumedTurn;
  • resume-ошибка (non-2xx / network throw): resumedTurn сброшен (следующий локальный ход флашит очередь);
  • resumed-ход: onFinish не флашит очередь (оба пути — plain и flush-on-abort); локальный send сбрасывает подавление;
  • здоровый resumed-финиш (message с видимым контентом): reconcile НЕ взводится, onResumeFallback НЕ зовётся, стор сохраняет полное сообщение (инвариант 9);
  • isDisconnect С видимым контентом: арм ЕСТЬ (reconcile + onResumeFallback(true)), restore НЕТ (стрипнутый снапшот не затирает более свежий частичный стрим);
  • пустой message (голодный реплей): restore + арм;
  • degraded-мердж: пока reconcileTailRef взведён, обновление initialRows мержит хвост; settled-хвост гасит флаг И зовёт onResumeFallback(false); isStreaming гасит флаг и зовёт onResumeFallback(false);
  • «Send now» скрыт при resumedTurn, виден при локальном стриме;
  • handleStop абортит attach-контроллер и зовёт onServerStop.

Окно (ai-chat-window / хук): degraded-интервал 2500 ⇔ degradedPoll ∧ младше 10 мин — И НИЧЕГО БОЛЬШЕ; ошибки fetch НЕ гасят (тик продолжается при isError); гаснет по onResumeFallback(false), капу, смене чата.

DoD PR 2: tsc; npx vitest run src/features/ai-chat целиком; grep по run-polling|useAiChatRunQuery|observedRow|AI_CHAT_RUN_RQ_KEY|getAiChatRun|shouldObserveRun|stoppingRun в apps/client/src пуст.

Ожидаемое поведение (спецификация; тесты сверять с ней)

Случай Ожидание
Reload через 1с после send (окно begin→seed, user-хвост) open()-at-begin: entry уже есть → attach ждёт кадры → живой стрим с нулевого кадра.
Attach в момент финиша (стрипнувшая вкладка) ≤30s retention + expect=live + anchor совпал: полный реплей + end → onFinish → (здоровый) ничего не мержится.
Здоровый финиш resumed-стрима НИЧЕГО не взводится: стор держит полное стримленное сообщение; обновление кэша — через onTurnFinished; клоббер streaming-снапшотом запрещён (инвариант 9).
204 у стрипнувшей вкладки (overflow / begin-failure / после retention / anchor-мисматч) restore строки + degraded-poll (2.5s) до settle → терминальный merge (вкл. stopped-маркер). Гаснет на локальном send/settled; ошибки fetch НЕ гасят (переживает рестарт сервера); кап 10 мин.
204 у user-хвоста Без restore; invalidate + reconcile-арм + poll-арм. Ран стартовал в окне send→begin → очередной тик застаёт seed → degraded-следование до settle; рана нет → тики впустую до капа (0.4 rps, принято).
«Голодный» attach (0 кадров: abortEntry) / обрыв tee (isDisconnect, пустой message) resumed-onFinish: restore (при стрипе) + reconcile + poll до settle.
isDisconnect с частичным контентом (рестарт сервера на середине) Арм (reconcile+poll) БЕЗ restore — экранный частичный не откатывается к mount-time снапшоту; поллинг переживает рестарт (ошибки не гасят) и доводит до aborted-маркера после boot-sweep.
Кросс-рановый attach (N+1 финишировал, N+2 уже живой) anchor-мисматч → 204 → restore N+1-частичного + poll → сходится к settled N+1, N+2 подъедет обычным refetch'ем истории. Чужой транскрипт в стор не реплеится (инвариант 6).
Две вкладки, Stop из одной Стопнувшая: «Response stopped.», частичное остаётся (restore/recovery не взводятся — контент видим). Вторая: чистый финиш с частичным контентом, очередь НЕ флашится (инвариант 7).
Очереди на N вкладках, естественный финиш Флашит только стартер (не-resumed ход); 409-гонки нет.
Медленный клиент Кап 8MB (2× реплея): реплей-бёрст сам по себе кап не пробивает; патологический случай (1Mbps + очень болтливый ран) принят.
Флаг off open/bind/generateMessageId не зовутся, attach всегда 204, resume не гоняется, legacy-провод байт-в-байт.
Вкладка открыта ДО старта рана в другой вкладке Не подключается (mount-only) — известное ограничение, фаза 2 (ws-сигнал).

Память реестра (для оценки)

Worst-case footprint: (число одновременных активных ранов) × ≤4MB кадров + ≤8MB на каждого реально застрявшего подписчика + 30s retention-хвосты. Однопроцессный деплой, раны ограничены «один на чат» — практический потолок единицы-десятки MB.

Не входит в объём

Redis/BullMQ-реестр (интерфейс готов); resumable-stream; изменения POST /ai-chat/run//stop; миграции; public-share; «чистый финиш» у второй вкладки при Stop (паритет); mainline-баг sendNow-без-server-stop у стартера (п. 6 правил); composer, активный в 1-3с окно установления attach (паритет, строгое улучшение против целого рана в мейнлайне).

Связанное

  • #184 — большой proposal (это фаза 1.5: транспорт).
  • PR #234 — фаза 1 (durable detached runs).
# Зачем PR #234 (фаза 1 #184) сделал раны durable: сервер доводит ход после дисконнекта браузера. Но UI-доставка осталась **двухпутёвой**: вкладка-стартер видит живой SSE-стрим, а переоткрытая вкладка — 2-секундный поллинг снапшотов `POST /ai-chat/run` без токен-стрима. Вся latch-механика F4/F5/F7 (`run-polling.ts`, `ai-chat-window.tsx`) существует только потому, что два транспорта дерутся за вью. **Цель:** один транспорт для всех вкладок. Ран публикует свой SSE-стрим в серверный реестр; любая вкладка подключается через `GET`-attach с реплеем кадров и живым хвостом. Наблюдатель становится обычным стримером (`status === 'streaming'`) → Stop-кнопка, прыгающие точки и инвалидации работают штатными путями. Поллинг и латчи удаляются. # ⚠️ Правила для исполнителя — прочитать до начала 1. **Все решения приняты.** Не отклоняться, не «улучшать». Невозможно сделать как написано / противоречие — СТОП и вопрос комментарием в ишью. 2. **Два PR строго по порядку**: PR 1 (сервер, «спящий»), потом PR 2 (клиент + удаление). Не смешивать. 3. Комментарии в коде — только на английском. Точечные правки. 4. **Не трогать**: legacy-путь (флаг off — байт-в-байт; `generateMessageId` поэтому гейтится на `runId`), семантику `POST /ai-chat/stream`, `POST /ai-chat/run`, `POST /ai-chat/stop`, public-share чат, схему БД (миграций нет). 5. **Не добавлять зависимостей** (всё есть в установленном AI SDK: server `ai@6.0.134` патченный, client `ai@6.0.207` + `@ai-sdk/react@3.0.209`). Никакого resumable-stream/Redis/BullMQ. 6. **Известный смежный баг мейнлайна — НЕ чинить здесь:** «Send now» у стартера в autonomous-режиме не делает server-stop → флаш может словить 409. Отдельная задача. # Проверенные факты об SDK (сверены по установленным исходникам — опираться, не перепроверять) - `pipeUIMessageStreamToResponse(res, { consumeSseStream })` тее-ит поток SSE-строк **до** сокета (server dist:5911-5920); tee-ветка живёт при смерти клиентского сокета. Кадры `data: {json}\n\n`, терминальный `data: [DONE]\n\n`. Heartbeat'ы (`: ping`) в tee не попадают. - `reconnectToStream`: **GET** `${api}/${chatId}/stream`, `chatId` = **store id** useChat (не серверный!), **без AbortSignal** (client dist:13334-13372). `prepareReconnectToStreamRequest` полностью заменяет URL. **Только 204 → null** (тихий no-op, статус не меняется, `onFinish` НЕ зовётся); другой не-2xx бросает → `onError` + красный баннер (и `onFinish` тоже НЕ зовётся — см. задачу 2.3 п. 3). - `resumeStream()` — mount-only эффект (react dist:221-225). Терминал resumed-стрима зовёт `onFinish` из `finally` (dist:13757-13766) — как у локального. - **Сид-стрип обязателен**: `text-start` всегда пушит новую парту (dist:5746-5754) — реплей поверх засеянной in-progress строки дублирует текст. То же верно для реплея ЗАВЕРШЁННОГО рана поверх settled-строки — поэтому settled-хвост никогда не резюмим. - Аборт: чанк `{"type":"abort"}` → `[DONE]` → конец. Клиентский процессор `abort`-чанк игнорирует → у вкладки, не нажимавшей Stop, это чистый финиш (`isAbort:false`). Принято как паритет — но см. задачу 2.3 п. 8 (подавление флаша). Error-чанк, напротив, БРОСАЕТ в процессоре (dist:6109-6111 → 13734-13736) → `isError` + баннер. - Локальный `stop()` НЕ абортит resumed-фетч (сигнал не проброшен), но взводит `isAbort=true` на будущий `onFinish` (dist:13686-13688) — источник «зомби»-ловушек, закрытых ниже. - `assistantId` присваивается awaited-вставкой (`ai-chat.service.ts:842-853`) ДО pipe-вызова (:1176) на обоих путях. `generateMessageId` без `originalMessages` штампует `start.messageId` (dist:5819-5834). Chat id — UUIDv7, `ParseUUIDPipe` по умолчанию (`all`) его принимает. - Порядок в `AiChatService.stream`: `beginRun` :574-580 → вставка user-строки :609 → сид ассистент-строки :842 (между ними — загрузка истории, настройки, **MCP-хендшейки** — секунды!) → pipe :1176. На этом порядке держатся open()-at-begin гарантии. - **Гонка refetch ↔ finalizeAssistant**: терминальная запись ассистент-строки — первый await внутри `onFinish` у streamText (:1026-1039), стартует примерно когда `[DONE]` уходит на провод. Инвалидация сразу после клиентского onFinish регулярно возвращает строку ещё `status:'streaming'` (со снапшотом последнего ЗАВЕРШЁННОГО шага, для одношагового ответа — возможно пустым сидом). Поэтому мержить streaming-снапшот поверх полного стримленного сообщения НЕЛЬЗЯ — п. 7 задачи 2.3 взводит recovery только когда мержить действительно нужно. - **TanStack v5**: `query.state.fetchFailureCount` СБРАСЫВАЕТСЯ в 0 в начале каждого фетча (query-core dist:353-358, 408-412) — межтиковые consecutive-ошибки им НЕ выразить. Поэтому degraded-poll НЕ гасится по ошибкам вообще (см. задачу 2.4): единственные границы — 10-мин кап, settled, локальный стрим, смена чата. Бонус: поллинг переживает рестарт сервера. # ИНВАРИАНТЫ (нарушение любого = сломанный дизайн; на каждый есть тест) 1. `bind`/`abortEntry` мутируют entry **только** при совпадении `runId`. 2. Retention-делит удаляет **только** свою entry: `if (entries.get(chatId) === entry) entries.delete(chatId)`. 3. `open()` при замене завершает старую entry **зеркально done-пути** (started-подписчикам `onEnd()` И удалить их; paused — `pendingEnd=true`; снять её таймер) — включая ещё «живую» (ран settled, tee-петля дочитывает). 4. Снапшот реплея и регистрация подписчика — в **одном синхронном блоке** внутри `attach` (paused-режим); между ними нет await. 5. В контроллере cleanup (`close`-хендлер → unsubscribe) регистрируется **до первой записи** в сокет, плюс явная проверка `req.raw.destroyed` после регистрации (close мог прийти во время await'ов); реплей+`start()` — в try/catch с unsubscribe. 6. `finished`-entry при `expect=live` отдаётся БЕЗ регистрации подписчика; без `expect=live` — 204. При `expect=live` с `anchor` несовпадение `entry.assistantMessageId !== anchor` → 204 (кросс-рановый реплей запрещён). 7. Resumed-ход никогда не флашит очередь (ни plain-flush, ни flush-on-abort). 8. Degraded-poll гаснет ТОЛЬКО по четырём причинам: локальный стрим (тред зовёт `onResumeFallback(false)`), settled-merge (тред зовёт `onResumeFallback(false)`), смена чата, 10-мин кап. Ошибки fetch поллинг НЕ гасят. 9. Здоровый финиш resumed-стрима НИЧЕГО не мержит из кэша в стор; restore стрипнутой строки — ТОЛЬКО при отсутствии видимого контента в пришедшем message (клоббер более свежего частичного стрима mount-time снапшотом запрещён — в т.ч. на `isDisconnect`). --- # PR 1 — сервер ## Задача 1.1. `apps/server/src/core/ai-chat/ai-chat-stream-registry.service.ts` (новый) ```ts export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000; export const RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024; // 2x the replay cap: a just-written 4MB replay burst alone can never trip the // per-subscriber cap (see controller); only a genuinely stalled socket can. export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES; export interface RunStreamCallbacks { onFrame: (frame: string) => void; onEnd: () => void; } export interface RunStreamAttachment { replay: string[]; finished: boolean; start(): void; // drain pending frames (order preserved) and go live unsubscribe(): void; // safe to call at any point, idempotent } interface Subscriber extends RunStreamCallbacks { started: boolean; pending: string[]; pendingEnd: boolean; } interface Entry { runId: string; // The persisted assistant row id of this run (set at bind; undefined if the // seed failed). Used by the attach anchor check (invariant 6). assistantMessageId?: string; frames: string[]; bytes: number; overflowed: boolean; finished: boolean; subscribers: Set<Subscriber>; retainTimer?: NodeJS.Timeout; } @Injectable() export class AiChatStreamRegistryService implements OnModuleDestroy { private readonly entries = new Map<string, Entry>(); // key: chatId open(chatId: string, runId: string): void; bind(chatId: string, runId: string, assistantMessageId: string | undefined, stream: ReadableStream<string>): void; abortEntry(chatId: string, runId: string): void; attach(chatId: string, expectLive: boolean, anchor: string | undefined, cb: RunStreamCallbacks): Promise<RunStreamAttachment | null>; // async: phase-2 Redis seam onModuleDestroy(): void; // clear all retain timers } ``` Семантика (реализовать ровно так): - **`open`** (в момент старта рана, до любых кадров): если по `chatId` есть старая entry — завершить её ЗЕРКАЛЬНО done-пути (инвариант 3: `finished=true`, started-подписчикам `onEnd()` и удалить их из Set, paused — `pendingEnd=true`, снять её retainTimer; поздний `done` её tee-петли сработает по замкнутой ссылке и не тронет новую благодаря identity-проверкам; повторный `onEnd` невозможен — started уже удалены), затем `entries.set(chatId, { runId, frames: [], bytes: 0, overflowed: false, finished: false, subscribers: new Set() })`. - **`bind`** (из `consumeSseStream`): найти entry; если нет или `entry.runId !== runId` — warn-лог и выход (инвариант 1). Иначе записать `entry.assistantMessageId = assistantMessageId` и fire-and-forget цикл `reader.read()`: - кадр: `frames.push(value); bytes += Buffer.byteLength(value);` затем если `bytes > RUN_STREAM_MAX_BUFFER_BYTES` → `overflowed = true; frames = [];` (один warn-лог; пересёкший кадр УЧТЁН и разослан; после overflow пуши в frames прекращаются, рассылка живым продолжается). Рассылка: каждому подписчику — started → `try { sub.onFrame(frame) } catch { subscribers.delete(sub) }`; paused → `sub.pending.push(frame)`. - `done` ИЛИ ошибка чтения: `finished = true`; started-подписчикам `onEnd()` и удалить; paused — `pendingEnd = true` (их дренирует `start()`); взвести retention-таймер с identity-проверкой (инвариант 2), `.unref()`. - **`abortEntry`** (из внешнего catch стрим-метода): identity-проверка по runId; если entry ещё не finished — тот же терминальный путь, что у `done`. - **`attach`** — синхронно, без await между шагами: entry нет ИЛИ `overflowed` → `null`; `expectLive && anchor && entry.assistantMessageId !== anchor` → `null` (инвариант 6: стрипнувшая вкладка не должна получить реплей ЧУЖОГО рана — например, N+1 финишировал и N+2 уже открыт в суб-секундном окне между fetch истории и attach); `finished && !expectLive` → `null`; `finished && expectLive` → `{ replay: frames.slice(), finished: true, start: noop, unsubscribe: noop }` (подписчик НЕ регистрируется); иначе зарегистрировать paused-подписчика (`started:false, pending:[], pendingEnd:false`) и вернуть `{ replay: frames.slice(), finished: false, start, unsubscribe }`. `start()`: синхронно доставить `pending` по порядку через onFrame, затем `started = true`; если `pendingEnd` — `onEnd()` и удалить. `unsubscribe()`: удалить из Set (идемпотентно). - Примечание к anchor: если `bind` ещё не случился (окно begin→pipe), `entry.assistantMessageId` undefined → мисматч с любым anchor → 204 → клиентский restore+poll путь сходится сам. Это норма, не чинить. ## Задача 1.2. Модуль `ai-chat.module.ts`: `AiChatStreamRegistryService` в `providers`. ## Задача 1.3. Врезки в существующий код 1. **`ai-chat.controller.ts`**, begin-хук (замыкание `runHooks` ~:345-374): сразу после успешного `beginRun` (там, где известен `runId`) добавить `this.streamRegistry.open(chatId, runId);`. Реестр инжектится в контроллер. 2. **`ai-chat.service.ts`**, pipe-вызов (~:1176), добавить в options ровно два поля — ОБА гейтятся на `runId` (правило 4: legacy-путь при флаге off — байт-в-байт, включая отсутствие `start.messageId`; сид `assistantId` при этом безусловный, поэтому гейт только по `assistantId` менял бы legacy-провод): ```ts // #184 phase 1.5: run-wrapped mode only — the legacy path (flag off) stays // byte-for-byte identical, including the absence of start.messageId. ...(runId ? { // Tee the SSE frames into the run-stream registry so late tabs can // attach (replay + live tail). consumeSseStream: ({ stream }: { stream: ReadableStream<string> }) => this.streamRegistry.bind(chatId, runId, assistantId, stream), // Stamp the persisted assistant row's DB id onto the streamed message so // every tab renders the SAME id as the DB row (id-based reconciliation). // Seeding is best-effort: when it failed, let the client generate the id. ...(assistantId ? { generateMessageId: () => assistantId } : {}), } : {}), ``` 3. **`ai-chat.service.ts`**, внешний catch стрим-метода (~:1241, рядом с аварийным `onSettled`): `if (runId) this.streamRegistry.abortEntry(chatId, runId);` — иначе подписчики открытой entry зависнут навсегда. ## Задача 1.4. Attach-эндпоинт `ai-chat.controller.ts`, роут **`GET /api/ai-chat/runs/:chatId/stream`** (+ query `expect`, `anchor`). `@UseGuards`/`@Throttle` — та же комбинация, что у `stream`-эндпоинта, лимит `{ limit: 60, ttl: 60000 }`; `@SkipTransform()`. (Бакеты throttler'а раздельные по хендлерам — проверено, с 25/мин стрима не пересекается.) ```ts @Get('runs/:chatId/stream') async attachRunStream( @Param('chatId', new ParseUUIDPipe()) chatId: string, @Query('expect') expect: string | undefined, @Query('anchor') anchor: string | undefined, @Req() req: FastifyRequest, @Res() res: FastifyReply, @AuthUser() user: User, @AuthWorkspace() workspace: Workspace, ) { await this.assertOwnedChat(chatId, user, workspace); // same gate as getRun let stopHeartbeat: () => void = () => undefined; const attachment = await this.streamRegistry.attach( chatId, expect === 'live', anchor, { onFrame: (frame) => { // Backpressure guard: 2x the replay cap, so the initial replay burst // alone can never trip it; only a genuinely stalled socket can. try { if (res.raw.writableLength > SUBSCRIBER_MAX_BUFFERED_BYTES) { res.raw.destroy(); // 'close' fires -> unsubscribe below return; } if (!res.raw.writableEnded) res.raw.write(frame); } catch { res.raw.destroy(); } }, onEnd: () => { stopHeartbeat(); if (!res.raw.writableEnded) res.raw.end(); }, }, ); if (!attachment) { res.status(204).send(); // the ONLY "nothing to resume" signal the SDK accepts return; } res.hijack(); // Cleanup BEFORE any write (invariant 5): a torn-down socket must not orphan // a paused subscriber whose pending queue would buffer the whole run. req.raw.once('close', () => { attachment.unsubscribe(); stopHeartbeat(); }); // A close emitted DURING the awaits above was missed by the listener — check. // (Healthy pending GETs have req.raw.destroyed === false, so no false // positives; returning without end() is fine — the socket is gone.) if (req.raw.destroyed) { attachment.unsubscribe(); return; } res.raw.on('error', () => undefined); try { res.raw.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache', 'x-vercel-ai-ui-message-stream': 'v1', 'x-accel-buffering': 'no', // deliberately NO Connection/Keep-Alive (hop-by-hop; Safari/HTTP2) }); res.raw.flushHeaders?.(); for (const frame of attachment.replay) res.raw.write(frame); if (attachment.finished) { res.raw.end(); return; } stopHeartbeat = startSseHeartbeat(res.raw, 15_000); attachment.start(); // drain pending accumulated during replay, go live } catch { attachment.unsubscribe(); stopHeartbeat(); res.raw.destroy(); } } ``` ## Задача 1.5. Тесты PR 1 - **registry.spec** (unit): реплей в порядке поступления; поздний attach = полный префикс + живой хвост; paused-подписчик получает pending в порядке, затем live (кадры между attach и start не теряются и не переупорядочиваются); finished во время паузы → onEnd на start(); `finished`+`expect=live` → реплей без регистрации подписчика; `finished` без expect → null; anchor-мисматч при `expect=live` → null (и до bind — `assistantMessageId` undefined → null); совпадение → attach; overflow → attach null, живой подписчик продолжает получать (оракул: push→учёт→проверка, пересёкший кадр разослан); open() поверх retained entry завершает старую и снимает её таймер; open() поверх LIVE entry (ран settled, tee ещё читает): started-подписчики получают ровно ОДИН `onEnd` (удалены при open — поздний done её петли не дублирует вызов), поздний done не удаляет новую entry (identity, инвариант 3); retention удаляет только свою entry; bind/abortEntry с чужим runId — no-op; бросающий onFrame выкидывает только этого подписчика, ingest-цикл жив. - **controller.attach.spec** (по образцу `ai-chat.controller.run.spec.ts`): 403 чужой чат; 204 (нет entry / finished без expect=live / anchor-мисматч); реплей пишется; close до start() → unsubscribe; close во время await'ов (`req.raw.destroyed` к моменту регистрации) → unsubscribe, ничего не пишется; cap → destroy; finished-реплей завершает ответ сразу. - **int-spec** (`test/integration/ai-chat-attach.int-spec.ts`, по образцу stream-int-спека с `MockLanguageModelV3`): дисконнект стартера на середине → attach → кадры идентичны полной последовательности вкл. `[DONE]`; attach СРАЗУ после старта рана (до первого кадра) получает поток с нулевого кадра; `requestStop` → подписчик видит `{"type":"abort"}` + `[DONE]` + конец. - Расширить `ai-chat.service.spec.ts`: `consumeSseStream` И `generateMessageId` присутствуют ⇔ run-wrapped (легаси-опции pipe-вызова НЕ содержат ни того, ни другого); `generateMessageId` = id засеянной строки; abortEntry зовётся из внешнего catch. **DoD PR 1:** tsc + build; существующие тесты зелёные без правок (кроме `ai-chat.service.spec.ts`); новые зелёные; флаг off — байт-в-байт (вкл. отсутствие `start.messageId`). Флаг on со СТАРЫМ клиентом: единственное видимое изменение — `start.messageId` = DB-id строки; старый клиент к id не привязан (единственный потребитель — `mergeObservedMessage`, которому совпадение id только помогает); намеренно и допустимо. --- # PR 2 — клиент ## Задача 2.1. Типы `types/ai-chat.types.ts`: в `IAiChatMessageRow` добавить `status?: string;` (провод уже несёт — `baseFields` включают `status`). ## Задача 2.2. `utils/resume-helpers.ts` (новый) - `isStreamingTail(rows)` — хвост `role==='assistant' && status==='streaming'`. - `isSettledAssistantTail(rows)` — хвост-ассистент со `status !== 'streaming'`. - `seedRows(rows, strip)` — при `strip` вернуть без последней строки. - `mergeById(messages, incoming)` — перенос СУЩЕСТВУЮЩЕЙ `mergeObservedMessage` из `run-polling.ts` (логика не меняется). - Тесты: перенесённые кейсы merge + новые для обоих предикатов и seedRows. ## Задача 2.3. `components/chat-thread.tsx` Разрешён ровно **один новый проп**: `onResumeFallback?: (active: boolean) => void` (управление degraded-poll в окне, задача 2.4). 0. **Объявления (все новые ref'ы/стейт этого блока):** ```ts // resume machinery refs (#184 phase 1.5) const attachAbortRef = useRef<AbortController | null>(null); const reconcileTailRef = useRef(false); const noStreamHandledRef = useRef(false); const onNoActiveStreamRef = useRef<(() => void) | null>(null); const [resumedTurn, setResumedTurn] = useState(false); const resumedTurnRef = useRef(false); // Identity-stable pair setter (bare useState setter + ref write): it is closed // over by the transport useMemo([]), so it MUST NOT capture state. const setResumedTurnPair = useCallback((v: boolean) => { resumedTurnRef.current = v; setResumedTurn(v); }, []); ``` 1. **Гейтинг (mount-time, в ref'ах):** ```ts // Attempt resume for any non-settled tail: a streaming tail (strip + expect // live replay) or a user tail (the run may exist but its assistant row is not // seeded yet — attach to the pre-opened registry entry and wait for frames). // A settled assistant tail must NEVER resume: replaying a finished run into a // store that already contains its message duplicates parts (SDK text-start // always pushes a new part). const stripRef = useRef(chatId !== null && isStreamingTail(initialRows ?? [])); const attemptResumeRef = useRef( autonomousRunsEnabled === true && chatId !== null && !isSettledAssistantTail(initialRows ?? []), ); const strippedRowRef = useRef<IAiChatMessageRow | null>( stripRef.current ? (initialRows ?? [])[initialRows!.length - 1] : null, ); ``` 2. **Сид:** `seedRows(initialRows ?? [], attemptResumeRef.current && stripRef.current).map(rowToUiMessage)`. 3. **Транспорт** — два новых поля в существующем `useMemo`: ```ts prepareReconnectToStreamRequest: () => ({ // SDK default URL uses the useChat STORE id — always build from the real chat id. // ?expect=live&anchor=<row id> ONLY when we stripped a streaming tail: expect=live // is the only case where a finished-retained replay is safe (the row is stripped, // replay rebuilds it), and the anchor pins the replay to OUR run — a mismatching // (newer) run must 204 into the restore+poll path instead of replaying a foreign // transcript into this store. api: `/api/ai-chat/runs/${chatIdRef.current}/stream${ stripRef.current ? `?expect=live&anchor=${strippedRowRef.current!.id}` : '' }`, }), fetch: async (input: RequestInfo | URL, init: RequestInit = {}) => { if ((init.method ?? 'GET') !== 'GET') return fetch(input, init); // send path untouched // Reconnect GET: the SDK passes no AbortSignal, so wire our own controller // for observer Stop; surface the silent 204; and clear the resumedTurn flag // on ANY no-onFinish outcome (204, HTTP error, network throw) so it cannot // suppress the NEXT local turn's queue flush. const controller = new AbortController(); attachAbortRef.current = controller; try { const response = await fetch(input, { ...init, signal: controller.signal }); if (response.status === 204) onNoActiveStreamRef.current?.(); else if (!response.ok) setResumedTurnPair(false); // resume error: no onFinish will come return response; } catch (err) { setResumedTurnPair(false); // network failure: no onFinish will come throw err; } }, ``` 4. **Mount-эффект**: ```ts useEffect(() => { if (attemptResumeRef.current) { setResumedTurnPair(true); void resumeStream(); } // Mount-only by design; the parent remounts per chat via `key`. // eslint-disable-next-line react-hooks/exhaustive-deps }, []); ``` 5. **204-обработчик** (`onNoActiveStream`, one-shot через `noStreamHandledRef` — НЕ занулять `onNoActiveStreamRef`), **строго четыре части**: - (а) только если `strippedRowRef.current` — вернуть строку в стор через `mergeById` (user-tail 204 сюда НЕ заходит — не разыменовывать null!); - (б) безусловно — `reconcileTailRef.current = true` + инвалидация `AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current)`; - (в) безусловно — `onResumeFallback?.(true)`. Поллинг окна — тупой таймер с 10-мин капом (задача 2.4); тред гасит его сам через `onResumeFallback(false)` на settled/локальном стриме. На user-tail 204 без рана он дотикает до капа впустую (0.4 rps, принято); если ран стартовал в окне send→begin — очередной тик застанет seed и degraded-следование дойдёт до settle; - (г) `setResumedTurnPair(false)` (204 = onFinish не будет; иначе флаг подавил бы флаш СЛЕДУЮЩЕГО локального хода). 6. **Реконсиляция + degraded-мердж** — один эффект с deps ровно `[initialRows, isStreaming, setMessages]`: - при `isStreaming === true` → сбросить `reconcileTailRef`, вызвать `onResumeFallback?.(false)` (инвариант 8: локальный ход гасит и мердж, и поллинг), выход; - иначе, если `reconcileTailRef.current === true` и последняя строка `initialRows` — ассистент: `mergeById` её в стор НА КАЖДОЕ обновление `initialRows` (пока degraded-poll активен, это и есть живой пошаговый прогресс); если её `status !== 'streaming'` (settled) — сбросить `reconcileTailRef` И вызвать `onResumeFallback?.(false)` (терминальный merge сделан — гасим поллинг явно; окно само его не погасит, у него только временной кап). 7. **Recovery после голодного/оборванного resumed-финиша** (инвариант 9). В `onFinish`, используя `wasResumed` из п. 8; **арм и restore гейтятся ПО-РАЗНОМУ**: - **арм** (reconcile + инвалидация messages + `onResumeFallback?.(true)`): если `wasResumed && (isDisconnect || !assistantMessageHasVisibleContent(message))` (реюз утилиты из `utils/message-content.ts`); - **restore стрипнутой строки** (`mergeById(strippedRowRef.current)`): ТОЛЬКО если `wasResumed && !assistantMessageHasVisibleContent(message) && strippedRowRef.current`. На `isDisconnect` С видимым контентом restore ЗАПРЕЩЁН: живой стрим мог уйти далеко вперёд mount-time снапшота, и restore откатил бы контент на экране (клоббер, инвариант 9) — там достаточно арма: поллинг доведёт до настоящего терминала. Здоровый resumed-финиш НЕ взводит НИЧЕГО: стор уже держит полное стримленное сообщение, обновление кэша придёт через `onTurnFinished`, а немедленный refetch гонится с `finalizeAssistant` (см. факт в шапке). Provider-error mid-run: error-чанк бросает в процессоре → `isError` + баннер; при видимом контенте recovery не нужен, при пустом — условие арма и так истинно. 8. **Порядок в `onFinish` — строго** (инвариант 7): (1) `const wasResumed = resumedTurnRef.current;` и сразу `setResumedTurnPair(false)`; (2) recovery п. 7 (по `wasResumed`); (3) стандартные ветки — при `wasResumed === true` пропустить ОБЕ ветки отправки: plain `flushNext()` И ветку `flushOnAbortRef`; `onTurnFinished` и прочее — как обычно (возможная двойная инвалидация messages после п. 7 — намеренная и безвредная, НЕ «дедуплицировать»). Дополнительно флаг сбрасывается при любом ЛОКАЛЬНОМ send (все три call-site `sendMessage`: onSend, flushNext, sendNow — перед вызовом `setResumedTurnPair(false)`). 9. **«Send now» на resumed-ходе скрыт**: в JSX очереди кнопка interrupt-send не рендерится при `resumedTurn` (стейт из п. 0; остаётся только удаление). Причина: локальный `stop()` не абортит resumed-фетч (зомби: клик «проглочен», а `flushOnAbortRef` выстрелил бы спустя минуты на естественном финише); вариант «стопнуть сервер и слать» отвергнут из-за 409-гонки с асинхронным finalize. 10. **Stop**: в `handleStop` добавить первой строкой `attachAbortRef.current?.abort();`. Остальное не менять. (Наблюдатель, нажавший Stop: `isAbort` с частичным контентом → recovery по п. 7 не взводится, restore не делается — частичное остаётся + нотис «Response stopped.»; персистентный aborted-маркер появится при следующем открытии — паритет со стартером.) ## Задача 2.4. `components/ai-chat-window.tsx` — degraded-poll фолбэк (замена латчей) - Стейт `const [degradedPoll, setDegradedPoll] = useState(false);` + `armedAtRef` (штамп `Date.now()` при включении); сбрасывать при смене `activeChatId` (инвариант 8). - Проп `onResumeFallback={...}` на `<ChatThread>` (сеттер, штампующий armedAt при `true`). - Константа `DEGRADED_POLL_MAX_MS = 10 * 60_000`. - `useAiChatMessagesQuery` получает опцию `refetchInterval` (прокинуть через хук), callback-форма — НАМЕРЕННО ТУПАЯ: `degradedPoll === true && Date.now() - armedAtRef.current < DEGRADED_POLL_MAX_MS ? 2500 : false`. НИКАКИХ проверок ошибок (`fetchFailureCount` в TanStack v5 сбрасывается на каждом фетче и consecutive-ошибки не выражает — см. факт в шапке; поллинг обязан переживать рестарт сервера) и НИКАКИХ проверок хвоста (semantics живут в треде: settled/локальный стрим гасят через `onResumeFallback(false)`, п. 6 задачи 2.3). Временной кап — единственная страховка от вечного тика (застрявшая `'streaming'`-строка до boot-sweep; user-tail без рана). - Доставка контента в degraded-режиме — целиком на стороне ChatThread (задача 2.3 п. 6); окно только владеет таймером. ## Задача 2.5. Список удаления (точный) - `utils/run-polling.ts`, `utils/run-polling.test.ts` — удалить (merge-логика и её кейсы переехали в задаче 2.2). - `queries/ai-chat-query.ts` — удалить `useAiChatRunQuery`, `AI_CHAT_RUN_RQ_KEY`; `queries/ai-chat-run-query.test.tsx` — удалить. - `services/ai-chat-service.ts` — удалить `getAiChatRun` (⚠️ `stopRun` оставить). - `types/ai-chat.types.ts` — удалить `IAiChatRun`, `IAiChatRunResponse`. - `components/ai-chat-window.tsx` — удалить: `localStreaming`+`onStreamingChange`; латч `stoppingRun` + reset; `removeQueries` из `handleServerStop` (сжимается до `stopRun(chatId).catch(notify)`); блок `useAiChatRunQuery`; F7-эффект; `isObservingRun`/`observedRow`; терминальный эффект (`finalizedRunIdRef`); пропсы `observedRow`/`onStreamingChange`; мёртвые импорты. - `components/chat-thread.tsx` — удалить проп `observedRow` + observer-merge эффект; проп `onStreamingChange` + эффект. - `components/chat-thread.test.tsx` — удалить describe «observer run merge (#184)». - Серверный `POST /ai-chat/run` не трогать. ## Задача 2.6. Тесты PR 2 `chat-thread.test.tsx` (мок-харнесс уже ловит транспорт/useChat; добавить спай `resumeStream`): - `resumeStream` на маунте ⇔ флаг on ∧ chatId ∧ хвост НЕ settled-ассистент; сид стрипается только при streaming-хвосте; user-хвост — attempt без стрипа; - URL: `?expect=live&anchor=<id стрипнутой строки>` только при стрипе; без стрипа — без query; - 204 при user-хвосте: НЕТ краша, НЕТ restore, ЕСТЬ reconcile+invalidate, ЕСТЬ `onResumeFallback(true)`; - 204 при streaming-хвосте: restore + invalidate + `onResumeFallback(true)` + сброс resumedTurn; - resume-ошибка (non-2xx / network throw): `resumedTurn` сброшен (следующий локальный ход флашит очередь); - resumed-ход: `onFinish` не флашит очередь (оба пути — plain и flush-on-abort); локальный send сбрасывает подавление; - здоровый resumed-финиш (message с видимым контентом): reconcile НЕ взводится, `onResumeFallback` НЕ зовётся, стор сохраняет полное сообщение (инвариант 9); - isDisconnect С видимым контентом: арм ЕСТЬ (reconcile + `onResumeFallback(true)`), restore НЕТ (стрипнутый снапшот не затирает более свежий частичный стрим); - пустой message (голодный реплей): restore + арм; - degraded-мердж: пока `reconcileTailRef` взведён, обновление `initialRows` мержит хвост; settled-хвост гасит флаг И зовёт `onResumeFallback(false)`; `isStreaming` гасит флаг и зовёт `onResumeFallback(false)`; - «Send now» скрыт при resumedTurn, виден при локальном стриме; - `handleStop` абортит attach-контроллер и зовёт `onServerStop`. Окно (`ai-chat-window` / хук): degraded-интервал 2500 ⇔ `degradedPoll` ∧ младше 10 мин — И НИЧЕГО БОЛЬШЕ; ошибки fetch НЕ гасят (тик продолжается при `isError`); гаснет по `onResumeFallback(false)`, капу, смене чата. **DoD PR 2:** tsc; `npx vitest run src/features/ai-chat` целиком; grep по `run-polling|useAiChatRunQuery|observedRow|AI_CHAT_RUN_RQ_KEY|getAiChatRun|shouldObserveRun|stoppingRun` в `apps/client/src` пуст. --- # Ожидаемое поведение (спецификация; тесты сверять с ней) | Случай | Ожидание | |---|---| | Reload через 1с после send (окно begin→seed, user-хвост) | open()-at-begin: entry уже есть → attach ждёт кадры → живой стрим с нулевого кадра. | | Attach в момент финиша (стрипнувшая вкладка) | ≤30s retention + `expect=live` + anchor совпал: полный реплей + end → onFinish → (здоровый) ничего не мержится. | | Здоровый финиш resumed-стрима | НИЧЕГО не взводится: стор держит полное стримленное сообщение; обновление кэша — через onTurnFinished; клоббер streaming-снапшотом запрещён (инвариант 9). | | 204 у стрипнувшей вкладки (overflow / begin-failure / после retention / anchor-мисматч) | restore строки + degraded-poll (2.5s) до settle → терминальный merge (вкл. stopped-маркер). Гаснет на локальном send/settled; ошибки fetch НЕ гасят (переживает рестарт сервера); кап 10 мин. | | 204 у user-хвоста | Без restore; invalidate + reconcile-арм + poll-арм. Ран стартовал в окне send→begin → очередной тик застаёт seed → degraded-следование до settle; рана нет → тики впустую до капа (0.4 rps, принято). | | «Голодный» attach (0 кадров: abortEntry) / обрыв tee (isDisconnect, пустой message) | resumed-onFinish: restore (при стрипе) + reconcile + poll до settle. | | isDisconnect с частичным контентом (рестарт сервера на середине) | Арм (reconcile+poll) БЕЗ restore — экранный частичный не откатывается к mount-time снапшоту; поллинг переживает рестарт (ошибки не гасят) и доводит до aborted-маркера после boot-sweep. | | Кросс-рановый attach (N+1 финишировал, N+2 уже живой) | anchor-мисматч → 204 → restore N+1-частичного + poll → сходится к settled N+1, N+2 подъедет обычным refetch'ем истории. Чужой транскрипт в стор не реплеится (инвариант 6). | | Две вкладки, Stop из одной | Стопнувшая: «Response stopped.», частичное остаётся (restore/recovery не взводятся — контент видим). Вторая: чистый финиш с частичным контентом, очередь НЕ флашится (инвариант 7). | | Очереди на N вкладках, естественный финиш | Флашит только стартер (не-resumed ход); 409-гонки нет. | | Медленный клиент | Кап 8MB (2× реплея): реплей-бёрст сам по себе кап не пробивает; патологический случай (1Mbps + очень болтливый ран) принят. | | Флаг off | open/bind/generateMessageId не зовутся, attach всегда 204, resume не гоняется, legacy-провод байт-в-байт. | | Вкладка открыта ДО старта рана в другой вкладке | Не подключается (mount-only) — известное ограничение, фаза 2 (ws-сигнал). | # Память реестра (для оценки) Worst-case footprint: (число одновременных активных ранов) × ≤4MB кадров + ≤8MB на каждого реально застрявшего подписчика + 30s retention-хвосты. Однопроцессный деплой, раны ограничены «один на чат» — практический потолок единицы-десятки MB. # Не входит в объём Redis/BullMQ-реестр (интерфейс готов); resumable-stream; изменения `POST /ai-chat/run`/`/stop`; миграции; public-share; «чистый финиш» у второй вкладки при Stop (паритет); mainline-баг sendNow-без-server-stop у стартера (п. 6 правил); composer, активный в 1-3с окно установления attach (паритет, строгое улучшение против целого рана в мейнлайне). # Связанное - #184 — большой proposal (это фаза 1.5: транспорт). - PR #234 — фаза 1 (durable detached runs).
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
epic
Large multi-phase effort spanning many changes
 needs-human
эскалация: нужно решение человека
 review/approved
в последнем ревью нет открытых blocking-находок
 review/changes-requested
последнее ревью оставило открытые blocking-находки
 review/needs
head не ревьюился (head != reviewed_head)
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
agent_coder agent_qa agent_reviewer agent_vscode vvzvlad
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: vvzvlad/gitmost#381
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?