fix(streams): re-poll after a live batch before waiting for a wakeup

.changeset/streams-held-batch-wakeup.md

@@ -0,0 +1,5 @@
+---
+"agents": patch
+---
+
+Streams: a live `read`/`readBatches` no longer misses chunks appended while the consumer is still handling the previous batch. It re-polls before waiting for the next append, so a reader no longer stalls until a later append or the stream's end.

packages/agents/src/streams/streams.ts

@@ -350,6 +350,10 @@ export class Streams extends LifecycleCapability {
         continue;
       }
 
+      // The yield above was a suspension point: appends made while the
+      // consumer held the batch woke no waiter, so re-poll before sleeping.
+      if (rows.length > 0) continue;
+
       // Live tail: wait for the next append or settlement, then re-poll.
       // Wakeups carry no data, so there is nothing to buffer or dedupe.
       await this.#waitForWakeup(streamId, signal);

packages/agents/src/tests/streams/capability.test.ts

@@ -354,6 +354,33 @@ describe("Streams capability", () => {
     });
   });
 
+  it("an append while the consumer holds a live batch is not a lost wakeup", async () => {
+    const stub = env.StreamHarnessObject.getByName(crypto.randomUUID());
+    await runInDurableObject(stub, async (instance: StreamHarnessObject) => {
+      const stream = await instance.streams.open("held-batch");
+      const order: string[] = [];
+      const reading = (async () => {
+        for await (const batch of instance.streams.readBatches("held-batch")) {
+          order.push(`batch:${batch.map((c) => c.seq).join(",")}`);
+          // The producer appends while this consumer still holds the batch
+          // (the generator is suspended at its yield, with no waiter).
+          if (batch[0].seq === 0) stream.append({ i: 1 });
+        }
+      })();
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      stream.append({ i: 0 });
+
+      // The held-batch chunk must arrive with NO further append or close.
+      const deadline = Date.now() + 2_000;
+      while (!order.includes("batch:1") && Date.now() < deadline) {
+        await new Promise((resolve) => setTimeout(resolve, 5));
+      }
+      expect(order).toEqual(["batch:0", "batch:1"]);
+      stream.close();
+      await reading;
+    });
+  });
+
   it("deleting a live stream wakes tailing readers", async () => {
     const stub = env.StreamHarnessObject.getByName(crypto.randomUUID());
     await runInDurableObject(stub, async (instance: StreamHarnessObject) => {