fix(chat): keep mailbox cursor behind pending input

Author: gtremper

docs/ai-chat/custom-agents.mdx

@@ -228,14 +228,15 @@ For full control, skip `createSession` and compose the primitives directly:
 | Method | Behavior |
 | --- | --- |
 | `peek()` | Return the buffer head when it is a message, without consuming it; otherwise return `undefined` |
-| `hasPending()` | Resolve `true` when any message is buffered; does not consume it |
+| `hasPending()` | Resolve `true` when the buffer head is a message; does not consume it |
 | `next({ timeoutInSeconds? })` | Consume exactly one message record in channel order, or resolve `undefined` when the optional timeout elapses |
 | `on(handler)` | Consume messages as they arrive and invoke the handler |
 | `waitWithIdleTimeout(options)` | Wait warm, then suspend the run until the next message arrives |
 
-`hasPending()` checks the local, already-delivered buffer. It does not query the
-remote Session channel or start a subscription. Use `waitWithIdleTimeout()` when
-the loop needs to idle until future input arrives.
+`hasPending()` checks whether the local, already-delivered buffer head is a
+message that `next()` can consume immediately. It does not query the remote
+Session channel or start a subscription. Use `waitWithIdleTimeout()` when the
+loop needs to idle until future input arrives.
 
 `next()` returns a readonly record envelope:
 
@@ -258,8 +259,9 @@ contrast, `on()` commits a record as soon as it dispatches the handler; avoid
 mixing `on()` and `next()` when a single loop owns mailbox consumption.
 
 The Session `.in` channel also carries control records such as handovers. If one
-comes before a message, `next()` leaves it for its own consumer and waits until
-that record has been handled.
+comes before a message, `hasPending()` stays `false` and `next()` leaves the
+control record for its own consumer. After that record is handled, the message
+becomes pending.
 
 A complete loop:
 

packages/core/src/v3/sessionStreams/index.ts

@@ -86,18 +86,6 @@ export class SessionStreamsAPI implements SessionStreamManager {
     return manager.peekRecord(sessionId, io);
   }
 
-  public peekRecordWhere(
-    sessionId: string,
-    io: SessionChannelIO,
-    predicate: SessionStreamRecordPredicate
-  ): SessionStreamRecord | undefined {
-    const manager = this.#getManager();
-    if (!manager.peekRecordWhere) {
-      throw new Error("The configured Session stream manager does not support selective records");
-    }
-    return manager.peekRecordWhere(sessionId, io, predicate);
-  }
-
   public lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
     return this.#getManager().lastSeqNum(sessionId, io);
   }

packages/core/src/v3/sessionStreams/manager.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
 import { StandardSessionStreamManager } from "./manager.js";
 import type { ApiClient } from "../apiClient/index.js";
 import type { SSEStreamPart } from "../apiClient/runStream.js";
+import { InputStreamTimeoutError } from "../inputStreams/types.js";
 
 // Single-shot mock that mimics S2's long-poll: delivers `records` once via
 // `onPart` on the first subscribe call, then keeps the returned async
@@ -44,6 +45,31 @@ function singleShotApiClient(
   } as unknown as ApiClient;
 }
 
+function repeatingApiClient(record: {
+  id: string;
+  recordId?: string;
+  chunk: unknown;
+  timestamp: number;
+}): ApiClient {
+  return {
+    async subscribeToSessionStream<T>(
+      _sessionIdOrExternalId: string,
+      _io: "out" | "in",
+      options?: { onPart?: (part: SSEStreamPart<T>) => void; signal?: AbortSignal }
+    ) {
+      options?.onPart?.(record as SSEStreamPart<T>);
+      const signal = options?.signal;
+      // eslint-disable-next-line require-yield
+      return (async function* () {
+        if (signal?.aborted) return;
+        await new Promise<void>((resolve) => {
+          signal?.addEventListener("abort", () => resolve(), { once: true });
+        });
+      })() as unknown as Awaited<ReturnType<ApiClient["subscribeToSessionStream"]>>;
+    },
+  } as unknown as ApiClient;
+}
+
 describe("StandardSessionStreamManager — minTimestamp filter", () => {
   const sessionId = "session-1";
   const io = "in" as const;
@@ -210,24 +236,39 @@ describe("StandardSessionStreamManager — record metadata", () => {
   });
 
   it("returns the same envelope when a record is redelivered", async () => {
-    const firstDelivery = new StandardSessionStreamManager(
-      singleShotApiClient([records[0]!]),
+    const manager = new StandardSessionStreamManager(
+      repeatingApiClient(records[0]!),
       "http://localhost"
     );
-    const redelivery = new StandardSessionStreamManager(
-      singleShotApiClient([records[0]!]),
+
+    const first = await manager.onceRecord(sessionId, io);
+    manager.disconnectStream(sessionId, io);
+    const replayed = await manager.onceRecord(sessionId, io);
+
+    expect(first).toEqual(replayed);
+
+    manager.disconnectStream(sessionId, io);
+    manager.disconnect();
+  });
+
+  it("returns immediately when the timeout is zero", async () => {
+    const manager = new StandardSessionStreamManager(
+      {
+        subscribeToSessionStream: () => {
+          throw new Error("zero-timeout reads must not subscribe");
+        },
+      } as unknown as ApiClient,
       "http://localhost"
     );
 
-    const first = await firstDelivery.onceRecord(sessionId, io);
-    const replayed = await redelivery.onceRecord(sessionId, io);
+    const result = await manager.onceRecord(sessionId, io, { timeoutMs: 0 });
 
-    expect(first).toEqual(replayed);
+    expect(result.ok).toBe(false);
+    if (!result.ok) {
+      expect(result.error).toBeInstanceOf(InputStreamTimeoutError);
+    }
 
-    firstDelivery.disconnectStream(sessionId, io);
-    firstDelivery.disconnect();
-    redelivery.disconnectStream(sessionId, io);
-    redelivery.disconnect();
+    manager.disconnect();
   });
 
   it("does not consume a matching record past an earlier unmatched record", async () => {
@@ -256,10 +297,10 @@ describe("StandardSessionStreamManager — record metadata", () => {
       { timeoutMs: 200 }
     );
 
-    expect(manager.peekRecordWhere(sessionId, io, (record) => record.id === "message-1")).toEqual({
-      id: "message-1",
-      seqNum: 51,
-      data: { kind: "message", payload: { id: "u1" } },
+    expect(manager.peekRecord(sessionId, io)).toEqual({
+      id: "handover-1",
+      seqNum: 50,
+      data: { kind: "handover" },
     });
     expect(manager.lastDispatchedSeqNum(sessionId, io)).toBeUndefined();
 
@@ -281,4 +322,149 @@ describe("StandardSessionStreamManager — record metadata", () => {
     manager.disconnectStream(sessionId, io);
     manager.disconnect();
   });
+
+  it("keeps the persisted cursor behind each earlier buffered record", async () => {
+    const manager = new StandardSessionStreamManager(
+      singleShotApiClient([
+        {
+          id: "50",
+          recordId: "message-1",
+          chunk: { kind: "message", payload: { id: "u1" } },
+          timestamp: 1000,
+        },
+        {
+          id: "51",
+          recordId: "stop-1",
+          chunk: { kind: "stop" },
+          timestamp: 2000,
+        },
+        {
+          id: "52",
+          recordId: "message-2",
+          chunk: { kind: "message", payload: { id: "u2" } },
+          timestamp: 3000,
+        },
+        {
+          id: "53",
+          recordId: "stop-2",
+          chunk: { kind: "stop" },
+          timestamp: 4000,
+        },
+      ]),
+      "http://localhost"
+    );
+    let resolveStop!: () => void;
+    let remainingStops = 2;
+    const stopConsumed = new Promise<void>((resolve) => {
+      resolveStop = resolve;
+    });
+
+    manager.on(sessionId, io, (data) => {
+      if ((data as { kind?: string }).kind !== "stop") return;
+      remainingStops--;
+      if (remainingStops === 0) resolveStop();
+      return true;
+    });
+    await stopConsumed;
+
+    expect(manager.peekRecord(sessionId, io)).toEqual({
+      id: "message-1",
+      seqNum: 50,
+      data: { kind: "message", payload: { id: "u1" } },
+    });
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBe(49);
+
+    const firstMessage = await manager.onceRecord(sessionId, io);
+    expect(firstMessage.ok && firstMessage.output.id).toBe("message-1");
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBe(51);
+
+    const secondMessage = await manager.onceRecord(sessionId, io);
+    expect(secondMessage.ok && secondMessage.output.id).toBe("message-2");
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBe(53);
+
+    manager.disconnectStream(sessionId, io);
+    manager.disconnect();
+  });
+
+  it("retains cursor barriers when disconnect clears the buffer", async () => {
+    const manager = new StandardSessionStreamManager(
+      singleShotApiClient([
+        {
+          id: "50",
+          recordId: "message-1",
+          chunk: { kind: "message", payload: { id: "u1" } },
+          timestamp: 1000,
+        },
+        {
+          id: "51",
+          recordId: "stop-1",
+          chunk: { kind: "stop" },
+          timestamp: 2000,
+        },
+      ]),
+      "http://localhost"
+    );
+    let resolveStop!: () => void;
+    const stopConsumed = new Promise<void>((resolve) => {
+      resolveStop = resolve;
+    });
+
+    manager.on(sessionId, io, (data) => {
+      if ((data as { kind?: string }).kind !== "stop") return;
+      resolveStop();
+      return true;
+    });
+    await stopConsumed;
+
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBe(49);
+    manager.disconnectStream(sessionId, io);
+    expect(manager.peekRecord(sessionId, io)).toBeUndefined();
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBe(49);
+
+    manager.setLastDispatchedSeqNum(sessionId, io, 51);
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBe(49);
+
+    manager.reset();
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBeUndefined();
+  });
+
+  it("does not expose a negative cursor when sequence zero is buffered", async () => {
+    const manager = new StandardSessionStreamManager(
+      singleShotApiClient([
+        {
+          id: "0",
+          recordId: "message-0",
+          chunk: { kind: "message", payload: { id: "u0" } },
+          timestamp: 1000,
+        },
+        {
+          id: "1",
+          recordId: "stop-1",
+          chunk: { kind: "stop" },
+          timestamp: 2000,
+        },
+      ]),
+      "http://localhost"
+    );
+    let resolveStop!: () => void;
+    const stopConsumed = new Promise<void>((resolve) => {
+      resolveStop = resolve;
+    });
+
+    manager.on(sessionId, io, (data) => {
+      if ((data as { kind?: string }).kind !== "stop") return;
+      resolveStop();
+      return true;
+    });
+    await stopConsumed;
+
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBeUndefined();
+
+    const message = await manager.onceRecord(sessionId, io);
+    expect(message.ok && message.output.id).toBe("message-0");
+    expect(manager.lastDispatchedSeqNum(sessionId, io)).toBe(1);
+
+    manager.disconnectStream(sessionId, io);
+    manager.disconnect();
+  });
 });