// Characterization tests for the session event ordering path. These pin the// CURRENT behavior of the cursor gate, replay-cursor hydration, the// canonical/runtime event merge, the text-delta coalescer, batch replay, and// the resume subscription lifecycle — so the session-runtime-controller// consolidation can refactor against a fixed contract instead of live users.import assert from "node:assert/strict";import { readFileSync } from "node:fs";import test, { afterEach } from "node:test";// Fake timers: node:test's t.mock.timers is not implemented by bun's shim, so// the timer-driven tests below use bun:test's jest clock instead.import { jest } from "bun:test"; afterEach(() => {  jest.useRealTimers();  jest.setSystemTime(); // reset any pinned Date.now}); import {  mergeCanonicalAndRuntimeEvents,  replayCursorAfterRuntimeHydration,} from "@/features/agent/messages/helpers";import { foldSessionEvents } from "@/features/agent/runtime/pi-event-applier";import type { RuntimeLoggedEvent } from "@/features/agent/messages";import type {  RuntimeEventPayload,  RuntimeEventSubscription,  RuntimeSessionSummary,  RuntimeStatus,} from "@/features/agent/runtime/api";import {  acceptRuntimeSeq,  adoptExternalCursor,  commitRuntimeSeq,  reconnectAfter,} from "@/features/agent/runtime/runtime-cursor";import { createSessionRuntimeController } from "@/features/agent/runtime/session-runtime-controller";import { createEffectTextDeltaCoalescer as createTextDeltaCoalescer } from "@/features/agent/runtime/effect-coalescer";import type { Session, SessionId } from "@/features/agent/runtime/types"; const fixture = JSON.parse(  readFileSync(new URL("../fixtures/session-event-log.json", import.meta.url), "utf8"),) as {  canonical: Record<string, unknown>[];  runtimeTail: RuntimeLoggedEvent[];}; // ----- cursor gate (runtime-cursor.ts) ----- test("cursor gate passes seq-less payloads through without advancing", () => {  const cursor = adoptExternalCursor();  assert.deepEqual(acceptRuntimeSeq(cursor, undefined), { accept: true, cursor });  const at7 = adoptExternalCursor(7);  assert.deepEqual(acceptRuntimeSeq(at7, undefined), { accept: true, cursor: at7 });}); test("cursor gate rejects equal and stale seqs, accepts strictly newer", () => {  const at5 = adoptExternalCursor(5);  assert.deepEqual(acceptRuntimeSeq(at5, 5), { accept: false, cursor: at5 });  assert.deepEqual(acceptRuntimeSeq(at5, 4), { accept: false, cursor: at5 });  assert.equal(acceptRuntimeSeq(at5, 6).accept, true);  assert.equal(acceptRuntimeSeq(at5, 6).cursor.receivedSeq, 6);  // No persisted cursor behaves as 0.  assert.equal(acceptRuntimeSeq(adoptExternalCursor(), 1).accept, true);}); test("adopting an external cursor moves the in-memory gate backwards", () => {  // Deliberately non-monotonic: a lastEventSeq reset (new prompt on the same  // Pi session) or replay hydration must move the gate BACKWARDS, while  // acceptRuntimeSeq alone is monotonic.  const reset = adoptExternalCursor(0);  assert.equal(acceptRuntimeSeq(reset, 1).accept, true);  assert.deepEqual(adoptExternalCursor(undefined), {    receivedSeq: undefined,    committedSeq: undefined,  });  assert.equal(adoptExternalCursor(43).receivedSeq, 43);}); test("committed cursor lags received and reconnect resumes from received", () => {  const cursor = acceptRuntimeSeq(adoptExternalCursor(2), 5).cursor;  assert.equal(cursor.receivedSeq, 5);  assert.equal(cursor.committedSeq, 2);  // Reconnect must use the highest RECEIVED seq: an unflushed coalesced delta  // is still in memory, so replaying it would double-apply.  assert.equal(reconnectAfter(cursor), 5);  const committed = commitRuntimeSeq(cursor, 5);  assert.equal(committed.committedSeq, 5);  // Commit is monotonic.  assert.equal(commitRuntimeSeq(committed, 3).committedSeq, 5);  assert.equal(reconnectAfter(adoptExternalCursor()), 0);}); // ----- replay cursor after navigation hydration (session/helpers.ts) ----- test("replay hydration reattaches from the runtime cursor when the runtime matches the pi session", () => {  const pi = "pi-1";  // Matched runtime seeds the cursor regardless of activity — an idle matched  // runtime still holds the full event backlog, and starting at seq 0 would  // replay the whole history on top of the hydrated transcript (the  // reopened-old-session double-history bug).  assert.equal(    replayCursorAfterRuntimeHydration({ active: true, piSessionId: pi, eventSeq: 42 }, pi),    42,  );  assert.equal(    replayCursorAfterRuntimeHydration({ active: false, piSessionId: pi, eventSeq: 42 }, pi),    42,  );  // Active runtime that has not reported a pi session id keeps the historical  // behavior of being treated as this session's runtime.  assert.equal(    replayCursorAfterRuntimeHydration({ active: true, piSessionId: null, eventSeq: 42 }, pi),    42,  );  // An idle unclaimed runtime or a runtime on a different pi session is not  // provably ours — do not adopt its cursor.  assert.equal(    replayCursorAfterRuntimeHydration({ active: false, piSessionId: null, eventSeq: 42 }, pi),    undefined,  );  assert.equal(    replayCursorAfterRuntimeHydration({ active: true, piSessionId: "pi-2", eventSeq: 42 }, pi),    undefined,  );  assert.equal(    replayCursorAfterRuntimeHydration({ active: true, piSessionId: pi, eventSeq: undefined }, pi),    undefined,  );  assert.equal(replayCursorAfterRuntimeHydration(null, pi), undefined);}); // ----- canonical + runtime merge (session/helpers.ts) ----- function userEvent(text: string): Record<string, unknown> {  return { type: "message", message: { role: "user", content: text } };} function assistantEvent(text: string): Record<string, unknown> {  return {    type: "message_end",    message: { role: "assistant", content: [{ type: "text", text }] },  };} test("merge cuts the canonical tail at the runtime's first user message", () => {  const canonical = [    userEvent("first question"),    assistantEvent("first answer"),    userEvent("second question"),    assistantEvent("settled second answer"),  ];  const runtime: RuntimeLoggedEvent[] = [    { seq: 1, event: userEvent("second question") },    { seq: 2, event: assistantEvent("live second answer") },  ];   const merged = mergeCanonicalAndRuntimeEvents(canonical, runtime);  // Canonical keeps only the first turn; the runtime owns the covered tail —  // the settled copy of the second turn must NOT appear (duplicate-bubble bug).  assert.deepEqual(merged, [    userEvent("first question"),    assistantEvent("first answer"),    userEvent("second question"),    assistantEvent("live second answer"),  ]);}); test("merge dedupes identical events and sorts the runtime log by seq", () => {  const canonical = [userEvent("only question")];  const runtime: RuntimeLoggedEvent[] = [    { seq: 2, event: assistantEvent("answer") },    { seq: 1, event: userEvent("only question") },    { seq: 3, event: assistantEvent("answer") },  ];   const merged = mergeCanonicalAndRuntimeEvents(canonical, runtime);  assert.deepEqual(merged, [userEvent("only question"), assistantEvent("answer")]);}); test("merge without runtime events returns the canonical log untouched", () => {  const canonical = [userEvent("q"), assistantEvent("a")];  assert.deepEqual(mergeCanonicalAndRuntimeEvents(canonical), canonical);}); test("merge aligns repeated prompts by the longest runtime message sequence", () => {  const canonical = [    userEvent("start"),    assistantEvent("started"),    userEvent("what happened"),    assistantEvent("first explanation"),    userEvent("what happened"),    assistantEvent("second explanation"),  ];  const runtime: RuntimeLoggedEvent[] = [    { seq: 2, event: assistantEvent("second explanation") },    { seq: 1, event: userEvent("what happened") },  ];   assert.deepEqual(mergeCanonicalAndRuntimeEvents(canonical, runtime), canonical);}); test("runtime replay preserves repeated user prompts across completed turns", () => {  const events = [    { type: "message_start", message: { role: "user", content: "what happened" } },    { type: "message_end", message: { role: "user", content: "what happened" } },    assistantEvent("first explanation"),    { type: "message_start", message: { role: "user", content: "what happened" } },    { type: "message_end", message: { role: "user", content: "what happened" } },    assistantEvent("second explanation"),  ];   const messages = foldSessionEvents(events).messages;   assert.deepEqual(    messages.map((message) => [message.role, message.text]),    [      ["user", "what happened"],      ["assistant", "first explanation"],      ["user", "what happened"],      ["assistant", "second explanation"],    ],  );}); // ----- batch replay over the golden event log (runtime/pi-event-applier.ts foldSessionEvents) ----- test("golden event log replays to the expected transcript", () => {  const { messages, title, startedAt, modelId } = foldSessionEvents(fixture.canonical);   assert.equal(title, "Summarize the GPU fleet");  assert.equal(startedAt, "2026-06-09T10:00:00.000Z");  assert.equal(modelId, "deepseek-v4-flash");   assert.deepEqual(    messages.map((message) => message.role),    ["user", "assistant", "assistant"],  );  assert.equal(messages[0]?.text, "Summarize the GPU fleet");   const toolTurn = messages[1];  assert.deepEqual(    (toolTurn?.blocks ?? []).map((block) => block.kind),    ["thinking", "tool"],  );  const toolBlock = toolTurn?.blocks?.find((block) => block.kind === "tool");  assert.equal(toolBlock?.kind === "tool" && toolBlock.status, "done");  assert.equal(    toolBlock?.kind === "tool" && toolBlock.text,    "4x RTX PRO 6000 Blackwell + 1x RTX 3090",  );   assert.equal(    messages[2]?.text,    "The fleet has four Blackwell cards and one RTX 3090.\n\n- Blackwells capped at 275W.\n- 3090 capped at 150W.",  );}); // ----- text delta coalescer (text-delta-coalescer.ts) ----- type FrameHarness = {  callbacks: (() => void)[];  cancelled: number;  schedule: (callback: () => void) => { cancel: () => void };  runAll: () => void;}; function frameHarness(): FrameHarness {  const harness: FrameHarness = {    callbacks: [],    cancelled: 0,    schedule: (callback) => {      harness.callbacks.push(callback);      return { cancel: () => (harness.cancelled += 1) };    },    runAll: () => {      const pending = harness.callbacks.splice(0);      for (const callback of pending) callback();    },  };  return harness;} function deltaEvent(type: string, delta: string): Record<string, unknown> {  return { type: "message_update", assistantMessageEvent: { type, delta } };} function appliedDelta(event: Record<string, unknown>): unknown {  return (event.assistantMessageEvent as Record<string, unknown> | undefined)?.delta;} test("coalescer concatenates same-kind deltas including standalone newlines", () => {  const applied: Record<string, unknown>[] = [];  const frames = frameHarness();  const coalescer = createTextDeltaCoalescer({    applyPiEvent: (_sessionId, event) => applied.push(event),    scheduleFrame: frames.schedule,  });   coalescer.enqueuePiEvent("s-1", deltaEvent("text_delta", "Row 1"));  coalescer.enqueuePiEvent("s-1", deltaEvent("text_delta", "\n"));  coalescer.enqueuePiEvent("s-1", deltaEvent("text_delta", "Row 2"));  assert.equal(applied.length, 0);   frames.runAll();  // One merged event; dropping the standalone "\n" delta was the table/paragraph  // whitespace bug (d9ede391).  assert.equal(applied.length, 1);  assert.equal(appliedDelta(applied[0]), "Row 1\nRow 2");}); test("coalescer flushNow applies pending once and a stale frame is harmless", () => {  const applied: Record<string, unknown>[] = [];  const frames = frameHarness();  const coalescer = createTextDeltaCoalescer({    applyPiEvent: (_sessionId, event) => applied.push(event),    scheduleFrame: frames.schedule,  });   coalescer.enqueuePiEvent("s-1", deltaEvent("text_delta", "Hello"));  coalescer.flushNow("s-1");  assert.equal(applied.length, 1);  assert.equal(frames.cancelled, 1);   // A frame callback firing after the explicit flush must not double-apply.  frames.runAll();  assert.equal(applied.length, 1);}); test("coalescer discard drops a session's pending deltas without applying them", () => {  // Cursor epoch resets (new prompt, replay hydration) discard stale pending  // merges instead of flushing them into the new epoch's transcript.  const applied: Record<string, unknown>[] = [];  const frames = frameHarness();  const coalescer = createTextDeltaCoalescer({    applyPiEvent: (_sessionId, event) => applied.push(event),    scheduleFrame: frames.schedule,  });   coalescer.enqueuePiEvent("s-1", deltaEvent("text_delta", "lost"));  coalescer.discard("s-1");  frames.runAll();  assert.equal(applied.length, 0);  assert.equal(frames.cancelled, 1);}); // NOTE: the coalescer no longer tracks a per-bubble assistant id — targeting// resolves inside reduceSessionEvent at apply time, and every target-changing// event (a user echo, a settled message, agent_end) is a non-delta that the// controller flushes ahead of (see "agent_end clears the mid-stream redirect"// below for the end-to-end ordering guarantee). // ----- live attachment lifecycle (session-runtime-controller.ts) ----- type ControllerHarness = {  session: () => Session;  patchSession: (patch: (session: Session) => Session) => void;  subscribeCalls: { after: number; piSessionId: string | null | undefined }[];  order: string[];  frames: FrameHarness;  controller: ReturnType<typeof createSessionRuntimeController>;  emit: (payload: RuntimeEventPayload) => void;  fail: () => void;  close: () => void;}; function createControllerHarness(  options: {    lastEventSeq?: number;    status?: RuntimeStatus | null;    sessionStatus?: Session["status"];  } = {},): ControllerHarness {  let session: Session = {    id: "s-1",    piSessionId: "pi-1",    title: "New session",    messages: [],    status: options.sessionStatus ?? "running",    error: "",    input: "",    lastEventSeq: options.lastEventSeq,  };  const subscribeCalls: ControllerHarness["subscribeCalls"] = [];  const handlers: { onPayload: (payload: RuntimeEventPayload) => void; onError: () => void }[] = [];  const order: string[] = [];  const frames = frameHarness();   const controller = createSessionRuntimeController({    api: {      loadRuntimeStatus: async () => options.status ?? null,      subscribeRuntimeEvents: (_runtime, after, piSessionId, eventHandlers) => {        subscribeCalls.push({ after, piSessionId });        handlers.push(eventHandlers);        return { close: () => order.push("transport-close") };      },    },    scheduleFrame: frames.schedule,  });  controller.bind({    commit: (sessionId, patch) => {      if (sessionId !== session.id) return;      const prev = session;      session = patch(session);      if (session === prev) return;      if (prev.status !== "idle" && session.status === "idle") order.push("idle-patch");      const joinedBlocks = (entry: Session) =>        (entry.messages.at(-1)?.blocks ?? []).map((block) => block.text).join("");      if (joinedBlocks(session) !== joinedBlocks(prev))        order.push(`blocks:${joinedBlocks(session)}`);    },    getSession: (sessionId) => (sessionId === session.id ? session : undefined),    getSessions: () => [session],  });  controller.reconcile([session]);   return {    session: () => session,    patchSession: (patch) => {      session = patch(session);    },    subscribeCalls,    order,    frames,    controller,    emit: (payload) => handlers.at(-1)?.onPayload(payload),    fail: () => handlers.at(-1)?.onError(),    close: () => {      controller.closeAll();      controller.unbind();    },  };} function partialDeltaEvent(delta: string, full: string): Record<string, unknown> {  return {    type: "message_update",    assistantMessageEvent: {      type: "text_delta",      delta,      contentIndex: 0,      partial: { role: "assistant", content: [{ type: "text", text: full }] },    },  };} const settle = () => new Promise<void>((resolve) => setImmediate(resolve)); test("controller reconnects from the highest received seq, not the configured start", async () => {  jest.useFakeTimers();  const harness = createControllerHarness({ status: { active: true } });   harness.emit({ type: "pi", seq: 5, event: partialDeltaEvent("a", "a") });  harness.emit({ type: "pi", seq: 3, event: partialDeltaEvent("b", "b") });  harness.fail();  await settle();  jest.advanceTimersByTime(1_000);   assert.deepEqual(    harness.subscribeCalls.map((call) => call.after),    [0, 5],  );  harness.close();}); test("reopened old session attaches from the hydrated runtime cursor, not seq 0", () => {  // Regression: reopening an old session hydrates the full transcript from the  // canonical log while the idle runtime still holds its whole event backlog.  // If replay hydration does not seed the cursor, the first SSE subscribe asks  // for `after=0` and the server replays the entire backlog on top of the  // rendered history — the double-transcript bug.  const harness = createControllerHarness({ sessionStatus: "idle" });  assert.equal(harness.subscribeCalls.length, 0);   // loadAndReplay finished: canonical rendered, matched idle runtime at seq 42.  harness.controller.noteReplayHydrated("s-1", 42);   // The user prompts; the session goes live and reconcile opens the SSE.  harness.patchSession((session) => ({ ...session, status: "running" }));  harness.controller.reconcile([harness.session()]);   assert.deepEqual(    harness.subscribeCalls.map((call) => call.after),    [42],  );  harness.close();}); test("controller applies agent_end immediately and settles the session idle", () => {  const harness = createControllerHarness();   harness.emit({ type: "pi", seq: 1, event: { type: "agent_end" } });   assert.equal(harness.session().status, "idle");  assert.equal(harness.session().activeAssistantId, undefined);  // The controller ensured an assistant bubble before applying the event.  assert.equal(harness.session().messages.at(-1)?.role, "assistant");  assert.equal(harness.session().lastEventSeq, 1);  harness.close();}); 

Lines 1–500 of 1030. Load more