diff --git a/docs/tech-debt.md b/docs/tech-debt.md index 6381050b..dd08df8b 100644 --- a/docs/tech-debt.md +++ b/docs/tech-debt.md @@ -271,3 +271,18 @@ address there. Remove this item when Bot API exposes a plain-text projection of `rich_message` beside the blocks, or when eve upstream resolves the addressing before the update reaches the Bridge. Until then the workaround in a group is to reply to one of Iva's messages. + +## 16. Rollup stale-cursor workaround for vercel/eve#2461 + +`scripts/lib/rollup-stale-cursor.ts` and the drain/ownership checks in +`scripts/memory/rollup.ts` work around an open upstream bug +([vercel/eve#2461](https://github.com/vercel/eve/issues/2461)): on a resumed +session, eve's client `result()` reads from the saved stream cursor and stops at +the first turn boundary without correlating it with the message just sent. Once +the cursor lags, the nightly report is a replay of an old turn. + +The Iva-side workaround is two small layers around eve, not a second session +system: drain `stream({ follow: false })` before every send into the parked +session, and refuse a result whose `message.received` is not this Turn's prompt +(per-execution nonce, `sentNotBefore` at send time). Remove both when a released +eve correlates `result()` with the sent turn. diff --git a/scripts/coverage-policy.test.ts b/scripts/coverage-policy.test.ts index bd183dd3..12c78528 100644 --- a/scripts/coverage-policy.test.ts +++ b/scripts/coverage-policy.test.ts @@ -8,9 +8,9 @@ import test from "node:test"; import { fileURLToPath } from "node:url"; const ROOT = fileURLToPath(new URL("../", import.meta.url)); -const EXPECTED_PRODUCTION_COUNT = 232; +const EXPECTED_PRODUCTION_COUNT = 233; const EXPECTED_INVENTORY_SHA256 = - "86aabf4601bb6c5d5d5d270da189bbd2ff765c72a37b452f278c9cc23d1a1957"; + "364dc2a2fc67e8fcb465c059fe01b0b7d76e3d7b880287b514ed6adee28d50a0"; // Node's native include globs filter loaded modules; they do not load untouched files. // This test pins the exact production path inventory and a separately measured 26-path @@ -116,6 +116,9 @@ const EXPECTED_INVENTORY_SHA256 = // `scripts/lib/update-check.test.ts` reports it at 100% lines, 92.16% branches and 100% // functions, with `scripts/check-update.ts` at 92.64% lines beside it, so the blind spot // stays 26. +// The rollup stale-cursor workaround `scripts/lib/rollup-stale-cursor.ts` came next. +// Scoped coverage over `scripts/lib/rollup-stale-cursor.test.ts` reports it, so the +// blind spot stays 26. const MEASURED_UNREPORTED_BY_CATEGORY = { frameworkBoundaries: [ "agent/agent.ts", diff --git a/scripts/lib/rollup-stale-cursor.test.ts b/scripts/lib/rollup-stale-cursor.test.ts new file mode 100644 index 00000000..75dd6a32 --- /dev/null +++ b/scripts/lib/rollup-stale-cursor.test.ts @@ -0,0 +1,520 @@ +/* eslint-disable @typescript-eslint/no-floating-promises -- Node's test runner owns registration promises. */ +// eve 0.30.8: result() читает поток с сохранённого streamIndex и останавливается на +// первой границе хода, не сверяя её с только что отправленным сообщением (vercel/eve#2461). +import test from "node:test"; +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { type ClientSession } from "eve/client"; +import fc from "fast-check"; +import { + attachRollupNonce, + drainStreamBefore, + drainStreamToTail, + isOwnTurnResult, + sentNotBeforeIso, +} from "./rollup-stale-cursor.ts"; + +function asClientStream(session: { + stream(options?: { + follow: false; + signal?: AbortSignal; + }): AsyncIterable; +}): Pick { + return session as Pick; +} + +const HERE = dirname(fileURLToPath(import.meta.url)); +const ROLLUP_SRC = readFileSync(join(HERE, "../memory/rollup.ts"), "utf8"); + +interface StreamEvent { + readonly type: string; + readonly data?: { readonly message?: string; readonly finishReason?: string }; + readonly meta?: { readonly at: string }; +} + +function received(message: string, at: string): StreamEvent { + return { + type: "message.received", + data: { message }, + meta: { at }, + }; +} + +function completed(message: string, at: string): StreamEvent { + return { + type: "message.completed", + data: { message, finishReason: "stop" }, + meta: { at }, + }; +} + +function waiting(at: string): StreamEvent { + return { type: "session.waiting", meta: { at } }; +} + +function turn(prompt: string, report: string, at: string): StreamEvent[] { + return [received(prompt, at), completed(report, at), waiting(at)]; +} + +function isTurnBoundary(event: StreamEvent): boolean { + return ( + event.type === "session.completed" || + event.type === "session.failed" || + event.type === "session.waiting" + ); +} + +// Модель result() eve 0.30.8: collectTurnEvents с сохранённого курсора до первой границы. +function resultFromCursor( + stream: readonly StreamEvent[], + streamIndex: number, +): { + readonly events: StreamEvent[]; + readonly message: string | undefined; + readonly status: "completed" | "failed" | "waiting"; +} { + const events: StreamEvent[] = []; + for (let i = streamIndex; i < stream.length; i++) { + const event = stream[i]; + if (event === undefined) break; + events.push(event); + if (isTurnBoundary(event)) break; + } + let message: string | undefined; + let status: "completed" | "failed" | "waiting" = "completed"; + for (const event of events) { + if ( + event.type === "message.completed" && + event.data?.finishReason !== "tool-calls" + ) { + message = event.data?.message; + } + if (event.type === "session.waiting") status = "waiting"; + if (event.type === "session.failed") status = "failed"; + } + return { events, message, status }; +} + +// origin/main в scripts/memory/rollup.ts: после result() нет сверки с промптом — +// любой waiting-ход с непустым текстом уходит в Telegram. +function mainWouldDeliver(result: { + readonly status: string; + readonly message: string | undefined; +}): boolean { + return result.status !== "failed" && Boolean(result.message); +} + +// PR 204: точный текст промпта + meta.at не старше process start минус 60с. +function pr204Owns( + events: readonly StreamEvent[], + prompt: string, + sentNotBefore: string, +): boolean { + return events.some( + (event) => + event.type === "message.received" && + event.data?.message === prompt && + (event.meta?.at ?? "") >= sentNotBefore, + ); +} + +class FakeSession { + streamIndex: number; + readonly events: StreamEvent[]; + constructor(events: StreamEvent[], streamIndex: number) { + this.events = events; + this.streamIndex = streamIndex; + } + stream(options?: { + follow: false; + signal?: AbortSignal; + }): AsyncIterable { + if ((options ?? { follow: false }).follow !== false) + throw new Error("expected follow:false"); + const signal = options?.signal; + return { + [Symbol.asyncIterator]: () => { + let closed = false; + const stop = (): void => { + closed = true; + }; + signal?.addEventListener("abort", stop, { once: true }); + if (signal?.aborted) stop(); + return { + next: async () => { + await Promise.resolve(); + if (closed || signal?.aborted) { + return { done: true as const, value: undefined }; + } + if (this.streamIndex >= this.events.length) { + return { done: true as const, value: undefined }; + } + const value = this.events[this.streamIndex]; + this.streamIndex += 1; + return { done: false as const, value }; + }, + return: () => { + closed = true; + return Promise.resolve({ done: true as const, value: undefined }); + }, + }; + }, + }; + } +} + +const OLD_PROMPT = + "You are processing long-term memory. It is now 2026-08-19. Process the completed day 2026-08-18."; +const TONIGHT_PROMPT = + "You are processing long-term memory. It is now 2026-08-24. Process the completed day 2026-08-23."; +const OLD_REPORT = "Обработан день 2026-08-18"; +const TONIGHT_REPORT = "Обработан день 2026-08-23"; + +test("origin/main delivers a lagged-cursor result from a previous night", () => { + const stream = [ + ...turn(OLD_PROMPT, OLD_REPORT, "2026-08-19T04:01:00.000Z"), + ...turn(TONIGHT_PROMPT, TONIGHT_REPORT, "2026-08-24T04:01:00.000Z"), + ]; + // Курсор отстал на один ход: result() останавливается на первой границе и отдаёт старый отчёт. + const result = resultFromCursor(stream, 0); + assert.equal(result.status, "waiting"); + assert.equal(result.message, OLD_REPORT); + assert.equal( + mainWouldDeliver(result), + true, + "main has no ownership check, so the five-day-old report would be delivered", + ); +}); + +test("a lagged cursor result is not this Turn's result", () => { + const stream = [ + ...turn(OLD_PROMPT, OLD_REPORT, "2026-08-19T04:01:00.000Z"), + ...turn(TONIGHT_PROMPT, TONIGHT_REPORT, "2026-08-24T04:01:00.000Z"), + ]; + const result = resultFromCursor(stream, 0); + const tonight = attachRollupNonce(TONIGHT_PROMPT, "tonight"); + assert.equal( + isOwnTurnResult(result.events, { + prompt: tonight, + sentNotBefore: sentNotBeforeIso(Date.parse("2026-08-24T04:00:00.000Z")), + }), + false, + ); +}); + +test("isOwnTurnResult fails closed on malformed timestamps", () => { + const prompt = attachRollupNonce(TONIGHT_PROMPT, "tonight"); + const sentNotBefore = "2026-08-24T04:00:00.000Z"; + assert.equal( + isOwnTurnResult([received(prompt, "zzzz")], { + prompt, + sentNotBefore, + }), + false, + ); + assert.equal( + isOwnTurnResult([received(prompt, "2026-08-24T03:30:00-01:00")], { + prompt, + sentNotBefore, + }), + true, + ); + assert.equal( + isOwnTurnResult([received(prompt, "2026-08-24T04:30:00+05:00")], { + prompt, + sentNotBefore, + }), + false, + ); +}); + +test("property: ownership check never crashes on junk events", () => { + fc.assert( + fc.property( + fc.array(fc.anything(), { maxLength: 30 }), + (events: unknown[]) => { + assert.equal( + typeof isOwnTurnResult(events, { + prompt: "expected", + sentNotBefore: "2026-08-24T04:00:00.000Z", + }), + "boolean", + ); + }, + ), + { seed: 18_713, numRuns: 200 }, + ); +}); + +test("drainStreamToTail advances a lagged cursor to the tail before send", async () => { + const stream = [ + ...turn(OLD_PROMPT, OLD_REPORT, "2026-08-19T04:01:00.000Z"), + ...turn(TONIGHT_PROMPT, TONIGHT_REPORT, "2026-08-24T04:01:00.000Z"), + ]; + const session = new FakeSession(stream, 0); + await drainStreamToTail(asClientStream(session)); + assert.equal(session.streamIndex, stream.length); + const tonight = attachRollupNonce(TONIGHT_PROMPT, "tonight"); + const live = [ + ...stream, + ...turn(tonight, TONIGHT_REPORT, "2026-08-24T04:02:00.000Z"), + ]; + const result = resultFromCursor(live, session.streamIndex); + assert.equal(result.message, TONIGHT_REPORT); + assert.equal( + isOwnTurnResult(result.events, { + prompt: tonight, + sentNotBefore: "2026-08-24T04:00:00.000Z", + }), + true, + ); +}); + +test("drainStreamToTail swallows a stream error so send can still proceed", async () => { + const session = { + stream(options?: { + follow: false; + signal?: AbortSignal; + }): AsyncIterable { + if ((options ?? { follow: false }).follow !== false) + throw new Error("expected follow:false"); + throw new Error("stream unavailable"); + }, + }; + const errors: string[] = []; + await drainStreamToTail(session, (error) => errors.push(error.message)); + assert.deepEqual(errors, ["stream unavailable"]); +}); + +test("drainStreamToTail finishes when both next and return hang past the timeout", async () => { + const session = { + stream(options?: { + follow: false; + signal?: AbortSignal; + }): AsyncIterable { + if ((options ?? { follow: false }).follow !== false) + throw new Error("expected follow:false"); + const signal = options?.signal; + return { + [Symbol.asyncIterator]: () => ({ + next: () => + new Promise>((_resolve, reject) => { + const fail = (): void => { + reject( + signal?.reason instanceof Error + ? signal.reason + : new Error("aborted"), + ); + }; + if (signal?.aborted) { + fail(); + return; + } + signal?.addEventListener("abort", fail, { once: true }); + }), + return: () => new Promise>(() => {}), + }), + }; + }, + }; + const errors: string[] = []; + const started = Date.now(); + await drainStreamToTail(session, (error) => errors.push(error.message), 50); + const elapsed = Date.now() - started; + assert.ok( + elapsed < 1000, + `hung drain must finish within the timeout, took ${elapsed}ms`, + ); + assert.equal(errors.length, 1); + assert.match(errors[0] ?? "", /timed out/); +}); + +test("drainStreamToTail passes abort signal so a late next does not advance the cursor", async () => { + let hasSignal = false; + let streamIndex = 0; + let closed = false; + let pendingResolve: ((result: IteratorResult) => void) | undefined; + const session = { + stream(options?: { + follow: false; + signal?: AbortSignal; + }): AsyncIterable { + hasSignal = options?.signal !== undefined; + const signal = options?.signal; + return { + [Symbol.asyncIterator]: () => ({ + next: () => + new Promise>((resolve, reject) => { + pendingResolve = (result) => { + if (closed || signal?.aborted) { + resolve({ done: true, value: undefined }); + return; + } + streamIndex += 1; + resolve(result); + }; + const fail = (): void => { + closed = true; + reject( + signal?.reason instanceof Error + ? signal.reason + : new Error("aborted"), + ); + }; + if (signal?.aborted) { + fail(); + return; + } + signal?.addEventListener("abort", fail, { once: true }); + }), + return: () => { + closed = true; + return Promise.resolve({ done: true as const, value: undefined }); + }, + }), + }; + }, + }; + const errors: string[] = []; + await drainStreamToTail( + asClientStream(session), + (error) => errors.push(error.message), + 50, + ); + assert.equal(hasSignal, true); + assert.equal(errors.length, 1); + assert.match(errors[0] ?? "", /timed out/); + pendingResolve?.({ done: false, value: { type: "late" } }); + assert.equal(streamIndex, 0); +}); + +test("PR 204 still accepts a delayed previous Turn event; a Rollup nonce does not", () => { + const processStart = Date.parse("2026-08-24T04:00:00.000Z"); + const delayedAt = "2026-08-24T04:00:05.000Z"; + const previous = attachRollupNonce(TONIGHT_PROMPT, "previous"); + const tonight = attachRollupNonce(TONIGHT_PROMPT, "tonight"); + const delayedPrevious = received(previous, delayedAt); + const delayedSameText = received(TONIGHT_PROMPT, delayedAt); + + assert.equal( + pr204Owns( + [delayedSameText], + TONIGHT_PROMPT, + new Date(processStart - 60_000).toISOString(), + ), + true, + "PR 204: same-date prompt + 60s slack accepts a delayed previous Turn event", + ); + assert.equal( + isOwnTurnResult([delayedSameText], { + prompt: TONIGHT_PROMPT, + sentNotBefore: sentNotBeforeIso(processStart), + }), + true, + "time check without a nonce still accepts an event stamped after process start", + ); + assert.equal( + isOwnTurnResult([delayedPrevious], { + prompt: tonight, + sentNotBefore: sentNotBeforeIso(processStart), + }), + false, + ); + assert.equal( + isOwnTurnResult([received(tonight, delayedAt)], { + prompt: tonight, + sentNotBefore: sentNotBeforeIso(processStart), + }), + true, + ); +}); + +test("sentNotBefore is the send instant, without a 60s slack window", () => { + const processStart = Date.parse("2026-08-24T04:00:00.000Z"); + const tonight = attachRollupNonce(TONIGHT_PROMPT, "tonight"); + const earlierSameNight = received(tonight, "2026-08-24T03:59:50.000Z"); + assert.equal( + isOwnTurnResult([earlierSameNight], { + prompt: tonight, + sentNotBefore: sentNotBeforeIso(processStart), + }), + false, + ); + assert.equal( + pr204Owns( + [received(TONIGHT_PROMPT, "2026-08-24T03:59:50.000Z")], + TONIGHT_PROMPT, + new Date(processStart - 60_000).toISOString(), + ), + true, + ); +}); + +test("rollup.ts uses the shared pre-send drain and refuses a foreign result", () => { + assert.match(ROLLUP_SRC, /vercel\/eve#2461/); + assert.match(ROLLUP_SRC, /drainStreamBefore\(/); + assert.match(ROLLUP_SRC, /isOwnTurnResult\(/); + assert.match(ROLLUP_SRC, /attachRollupNonce\(/); + assert.match(ROLLUP_SRC, /sentNotBeforeIso\(/); + assert.match( + ROLLUP_SRC, + /const send = async \(\) => \{\s+const sentNotBefore = sentNotBeforeIso\(\);\s+return \{\s+response: await session\.send\(prompt\),\s+sentNotBefore,/, + ); + assert.doesNotMatch(ROLLUP_SRC, /Date\.now\(\) - 60_000/); + assert.doesNotMatch(ROLLUP_SRC, /drainBeforeSend/); + const ownAt = ROLLUP_SRC.indexOf("isOwnTurnResult("); + const saveAt = ROLLUP_SRC.indexOf( + "saveSession(session.state, sessionCreatedAt);", + ); + assert.ok( + ownAt > 0 && ownAt < saveAt, + "refuse stale before saving the cursor", + ); +}); + +test("the production pre-send helper drains before every send and a foreign result is refused", async () => { + const order: string[] = []; + const prompt = attachRollupNonce(TONIGHT_PROMPT, "tonight"); + const foreign = turn(OLD_PROMPT, OLD_REPORT, "2026-08-19T04:01:00.000Z"); + const session = { + stream(options?: { follow?: boolean; signal?: AbortSignal }) { + if (options?.follow !== false) throw new Error("expected follow:false"); + order.push("drain"); + return { + async *[Symbol.asyncIterator]() { + /* empty tail: nothing parked ahead of send */ + }, + }; + }, + send() { + order.push("send"); + return Promise.resolve(); + }, + result() { + order.push("result"); + return Promise.resolve({ events: resultFromCursor(foreign, 0).events }); + }, + }; + await drainStreamBefore(asClientStream(session), () => session.send()); + await drainStreamBefore(asClientStream(session), () => session.send()); + await drainStreamBefore(asClientStream(session), () => session.send()); + const result = await session.result(); + assert.deepEqual(order, [ + "drain", + "send", + "drain", + "send", + "drain", + "send", + "result", + ]); + assert.equal( + isOwnTurnResult(result.events, { + prompt, + sentNotBefore: sentNotBeforeIso(Date.parse("2026-08-24T04:00:00.000Z")), + }), + false, + ); +}); diff --git a/scripts/lib/rollup-stale-cursor.ts b/scripts/lib/rollup-stale-cursor.ts new file mode 100644 index 00000000..8f5077af --- /dev/null +++ b/scripts/lib/rollup-stale-cursor.ts @@ -0,0 +1,89 @@ +// Обход vercel/eve#2461: на резюмнутой сессии eve-клиент читает поток с сохранённого +// streamIndex и останавливается на первой границе хода, не сверяя её с только что +// отправленным сообщением. Отставший курсор превращает result() в чтение старого хода: +// ночной отчёт уходит пятидневной давности, а падение реального хода не всплывает. +// +// Два слоя, оба временные: дочитать stream({follow:false}) до хвоста перед send и +// отказаться от результата, в событиях которого нет нашего message.received. Промпт +// несёт одноразовый nonce, чтобы повтор той же даты не принял чужой ход; нижняя +// граница времени — момент send, без запасной минуты. Снять оба слоя, когда eve +// свяжет result() с отправленным ходом (vercel/eve#2461). + +import { type ClientSession } from "eve/client"; +import { DEFAULT_TURN_TIMEOUT_MS } from "./rollup-turn.ts"; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +export function attachRollupNonce(prompt: string, nonce: string): string { + return `${prompt}\n`; +} + +export function sentNotBeforeIso(nowMs: number = Date.now()): string { + return new Date(nowMs).toISOString(); +} + +export function isOwnTurnResult( + events: readonly unknown[], + { + prompt, + sentNotBefore, + }: { + readonly prompt: string; + readonly sentNotBefore: string; + }, +): boolean { + const sentAt = Date.parse(sentNotBefore); + if (!Number.isFinite(sentAt)) return false; + return events.some((event) => { + if (!isRecord(event) || event.type !== "message.received") return false; + if (!isRecord(event.data) || !isRecord(event.meta)) return false; + const receivedAt = + typeof event.meta.at === "string" ? Date.parse(event.meta.at) : NaN; + return ( + event.data.message === prompt && + Number.isFinite(receivedAt) && + receivedAt >= sentAt + ); + }); +} + +export async function drainStreamToTail( + session: Pick, + onError?: (error: Error) => void, + timeoutMs: number = DEFAULT_TURN_TIMEOUT_MS, +): Promise { + // Зависший bounded-read иначе остановит ночь до guardedTurn(). AbortSignal — + // контракт stream() у eve: for-await его достаточно. По таймауту abort сигнала + // отклоняет pending next(); return() не вызываем — без своей границы он сам + // может зависнуть. + const controller = new AbortController(); + const { signal } = controller; + const timer = setTimeout(() => { + controller.abort(new Error("pre-send stream drain timed out")); + }, timeoutMs); + try { + for await (const event of session.stream({ + follow: false, + signal, + })) { + void event; + } + } catch (error) { + onError?.(error instanceof Error ? error : new Error(String(error))); + } finally { + clearTimeout(timer); + } +} + +/** Drain the saved cursor before an action that starts the next Turn. */ +export async function drainStreamBefore( + session: Pick, + action: () => Promise, + onError?: (error: Error) => void, + timeoutMs: number = DEFAULT_TURN_TIMEOUT_MS, +): Promise { + await drainStreamToTail(session, onError, timeoutMs); + return await action(); +} diff --git a/scripts/memory/rollup.ts b/scripts/memory/rollup.ts index ccfd5003..76237061 100644 --- a/scripts/memory/rollup.ts +++ b/scripts/memory/rollup.ts @@ -7,6 +7,7 @@ // Requires: a running agent (eve start) and a vault to write into. The processing rules // (scripts/memory/instructions/) ship with the repo. Date is in ASSISTANT_TIMEZONE. import { appendFileSync, mkdirSync, readFileSync, rmSync } from "node:fs"; +import { randomUUID } from "node:crypto"; import { dirname, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { Client, type MessageResult, type SessionState } from "eve/client"; @@ -36,6 +37,12 @@ import { resolveTurnTimeoutMs, withTurnTimeout, } from "../lib/rollup-turn.ts"; +import { + attachRollupNonce, + drainStreamBefore, + isOwnTurnResult, + sentNotBeforeIso, +} from "../lib/rollup-stale-cursor.ts"; import { sendTelegramHtml } from "../lib/telegram-send.ts"; type Period = "daily" | "weekly" | "monthly" | "yearly"; @@ -289,16 +296,34 @@ const guardedTurn = ( ) => withTurnTimeout( async () => { - let response; + const send = async () => { + const sentNotBefore = sentNotBeforeIso(); + return { + response: await session.send(prompt), + sentNotBefore, + }; + }; + let sent: Awaited>; try { - response = await session.send(prompt); + sent = session.state.sessionId + ? await drainStreamBefore( + session, + send, + (error) => { + console.error( + `rollup ${period}: ${label}: pre-send stream drain failed (${error.message}) — continuing with current cursor`, + ); + }, + TURN_TIMEOUT_MS, + ) + : await send(); } catch (error) { onSendRejected(); throw error; } - const result = response.result(); + const result = sent.response.result(); onAccepted(result); - return await result; + return { result: await result, sentNotBefore: sent.sentNotBefore }; }, { timeoutMs: TURN_TIMEOUT_MS, label }, ); @@ -334,14 +359,19 @@ const coreBeforeTurn = period === "daily" ? readCoreText(CORE_PATH) : ""; const saved = loadSession(); let sessionCreatedAt = saved?.createdAt ?? Date.now(); let session = saved ? client.session(saved.state) : client.session(); -let result; +// Обход vercel/eve#2461: result() на резюмнутой сессии может вернуть чужой ход. +// Nonce делает промпт уникальным для этого Rollup; guardedTurn сдвигает курсор +// на хвост перед каждым send. Снять, когда eve свяжет result() с отправленным ходом. +const mainPrompt = attachRollupNonce(buildPrompt(period, today), randomUUID()); +let result: MessageResult; +let sentNotBefore: string; let accepted = false; let sendRejected = false; let acceptedTurnResult: Promise | undefined; try { - result = await guardedTurn( + ({ result, sentNotBefore } = await guardedTurn( session, - buildPrompt(period, today), + mainPrompt, "main-turn", (turnResult) => { accepted = true; @@ -350,7 +380,7 @@ try { () => { sendRejected = true; }, - ); + )); } catch (e) { // The parked session may be gone (iva reset quarantined the store) or hung on resume — // fall back to a fresh one once only after proving the old turn cannot keep writing. @@ -383,17 +413,34 @@ try { sessionCreatedAt = Date.now(); // Ровно одна попытка: второй сбой уходит наверх и роняет юнит с ненулевым кодом. try { - result = await guardedTurn( + ({ result, sentNotBefore } = await guardedTurn( session, - buildPrompt(period, today), + mainPrompt, "main-turn", - ); + )); } catch (retryError) { if ((retryError as { code?: string }).code === "ROLLUP_TURN_TIMEOUT") await cancelTurnQuietly(session); throw retryError; } } +if ( + !isOwnTurnResult(result.events, { + prompt: mainPrompt, + sentNotBefore, + }) +) { + console.error( + `rollup ${period}: result does not match the prompt just sent (stale stream cursor) — dropping session`, + ); + logAbandoned(session.state, "stale-result"); + try { + rmSync(SESSION_FILE, { force: true }); + } catch { + /* курсор — кэш, его потеря не должна ронять ночь */ + } + process.exit(1); +} saveSession(session.state, sessionCreatedAt); // An interactive turn ends with status "waiting" (the session is ready for the next message),